CDM-18 : as all defaults are now in KnownProperties, removing Utils.getSparkProp* methods in favour of PropertyHelper

mieslep · mieslep · commit 098c453de73d · 2023-03-28T12:12:38.000+01:00
diff --git a/src/main/java/datastax/astra/migrate/AbstractJobSession.java b/src/main/java/datastax/astra/migrate/AbstractJobSession.java
@@ -6,7 +6,6 @@
 import com.datastax.oss.driver.api.core.cql.Row;
 import com.datastax.oss.driver.shaded.guava.common.util.concurrent.RateLimiter;
 import datastax.astra.migrate.properties.KnownProperties;
-import org.apache.commons.lang.StringUtils;
 import org.apache.spark.SparkConf;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -35,57 +34,45 @@ protected AbstractJobSession(CqlSession originSession, CqlSession targetSession,
         this.originSessionSession = originSession;
         this.targetSession = targetSession;
 
-        batchSize = new Integer(Util.getSparkPropOr(sc, KnownProperties.SPARK_BATCH_SIZE, "5"));
-        fetchSizeInRows = new Integer(Util.getSparkPropOr(sc, KnownProperties.READ_FETCH_SIZE, "1000"));
-        printStatsAfter = new Integer(Util.getSparkPropOr(sc, KnownProperties.SPARK_STATS_AFTER, "100000"));
-        if (printStatsAfter < 1) {
-            printStatsAfter = 100000;
+        batchSize = propertyHelper.getInteger(KnownProperties.SPARK_BATCH_SIZE);
+        fetchSizeInRows = propertyHelper.getInteger(KnownProperties.READ_FETCH_SIZE);
+        printStatsAfter = propertyHelper.getInteger(KnownProperties.SPARK_STATS_AFTER);
+        if (!propertyHelper.meetsMinimum(KnownProperties.SPARK_STATS_AFTER, printStatsAfter, 1)) {
+            logger.warn(KnownProperties.SPARK_STATS_AFTER +" must be greater than 0.  Setting to default value of " + KnownProperties.getDefaultAsString(KnownProperties.SPARK_STATS_AFTER));
+            propertyHelper.setProperty(KnownProperties.SPARK_STATS_AFTER, KnownProperties.getDefault(KnownProperties.SPARK_STATS_AFTER));
+            printStatsAfter = propertyHelper.getInteger(KnownProperties.SPARK_STATS_AFTER);
         }
 
-        readLimiter = RateLimiter.create(new Integer(Util.getSparkPropOr(sc, KnownProperties.SPARK_LIMIT_READ, "20000")));
-        writeLimiter = RateLimiter.create(new Integer(Util.getSparkPropOr(sc, KnownProperties.SPARK_LIMIT_WRITE, "40000")));
-        maxRetries = Integer.parseInt(sc.get(KnownProperties.SPARK_MAX_RETRIES, "0"));
+        readLimiter = RateLimiter.create(propertyHelper.getInteger(KnownProperties.SPARK_LIMIT_READ));
+        writeLimiter = RateLimiter.create(propertyHelper.getInteger(KnownProperties.SPARK_LIMIT_WRITE));
+        maxRetries = propertyHelper.getInteger(KnownProperties.SPARK_MAX_RETRIES);
 
-        originKeyspaceTable = Util.getSparkProp(sc, KnownProperties.ORIGIN_KEYSPACE_TABLE);
-        targetKeyspaceTable = Util.getSparkProp(sc, KnownProperties.TARGET_KEYSPACE_TABLE);
+        originKeyspaceTable = propertyHelper.getString(KnownProperties.ORIGIN_KEYSPACE_TABLE);
+        targetKeyspaceTable = propertyHelper.getString(KnownProperties.TARGET_KEYSPACE_TABLE);
 
-        String ttlColsStr = Util.getSparkPropOrEmpty(sc, KnownProperties.ORIGIN_TTL_COLS);
+        String ttlColsStr = propertyHelper.getAsString(KnownProperties.ORIGIN_TTL_COLS);
         if (null != ttlColsStr && ttlColsStr.trim().length() > 0) {
             for (String ttlCol : ttlColsStr.split(",")) {
                 ttlCols.add(Integer.parseInt(ttlCol));
             }
         }
 
-        String writeTimestampColsStr = Util.getSparkPropOrEmpty(sc, KnownProperties.ORIGIN_WRITETIME_COLS);
+        String writeTimestampColsStr = propertyHelper.getAsString(KnownProperties.ORIGIN_WRITETIME_COLS);
         if (null != writeTimestampColsStr && writeTimestampColsStr.trim().length() > 0) {
             for (String writeTimeStampCol : writeTimestampColsStr.split(",")) {
                 writeTimeStampCols.add(Integer.parseInt(writeTimeStampCol));
             }
         }
 
-        writeTimeStampFilter = Boolean
-                .parseBoolean(Util.getSparkPropOr(sc, KnownProperties.ORIGIN_FILTER_WRITETS_ENABLED, "false"));
+        writeTimeStampFilter = propertyHelper.getBoolean(KnownProperties.ORIGIN_FILTER_WRITETS_ENABLED);
         // batchsize set to 1 if there is a writeFilter
         if (writeTimeStampFilter) {
             batchSize = 1;
         }
 
-        String minWriteTimeStampFilterStr =
-                Util.getSparkPropOr(sc, KnownProperties.ORIGIN_FILTER_WRITETS_MIN, "0");
-        if (null != minWriteTimeStampFilterStr && minWriteTimeStampFilterStr.trim().length() > 1) {
-            minWriteTimeStampFilter = Long.parseLong(minWriteTimeStampFilterStr);
-        }
-        String maxWriteTimeStampFilterStr =
-                Util.getSparkPropOr(sc, KnownProperties.ORIGIN_FILTER_WRITETS_MAX, "0");
-        if (null != maxWriteTimeStampFilterStr && maxWriteTimeStampFilterStr.trim().length() > 1) {
-            maxWriteTimeStampFilter = Long.parseLong(maxWriteTimeStampFilterStr);
-        }
-
-        String customWriteTimeStr =
-                Util.getSparkPropOr(sc, KnownProperties.TARGET_CUSTOM_WRITETIME, "0");
-        if (null != customWriteTimeStr && customWriteTimeStr.trim().length() > 1 && StringUtils.isNumeric(customWriteTimeStr.trim())) {
-            customWritetime = Long.parseLong(customWriteTimeStr);
-        }
+        minWriteTimeStampFilter = propertyHelper.getLong(KnownProperties.ORIGIN_FILTER_WRITETS_MIN);
+        maxWriteTimeStampFilter = propertyHelper.getLong(KnownProperties.ORIGIN_FILTER_WRITETS_MAX);
+        customWritetime = propertyHelper.getLong(KnownProperties.TARGET_CUSTOM_WRITETIME);
 
         logger.info("PARAM -- Read Consistency: {}", readConsistencyLevel);
         logger.info("PARAM -- Write Consistency: {}", writeConsistencyLevel);
@@ -106,9 +93,9 @@ protected AbstractJobSession(CqlSession originSession, CqlSession targetSession,
                     Instant.ofEpochMilli(maxWriteTimeStampFilter / 1000));
         }
 
-        String selectCols = Util.getSparkProp(sc, KnownProperties.ORIGIN_COLUMN_NAMES);
-        String partitionKey = Util.getSparkProp(sc, KnownProperties.ORIGIN_PARTITION_KEY);
-        String originSelectCondition = Util.getSparkPropOrEmpty(sc, KnownProperties.ORIGIN_FILTER_CONDITION);
+        String selectCols = propertyHelper.getAsString(KnownProperties.ORIGIN_COLUMN_NAMES);
+        String partitionKey = propertyHelper.getAsString(KnownProperties.ORIGIN_PARTITION_KEY);
+        String originSelectCondition = propertyHelper.getAsString(KnownProperties.ORIGIN_FILTER_CONDITION);
         if (!originSelectCondition.isEmpty() && !originSelectCondition.trim().toUpperCase().startsWith("AND")) {
             originSelectCondition = " AND " + originSelectCondition;
         }
@@ -121,11 +108,11 @@ protected AbstractJobSession(CqlSession originSession, CqlSession targetSession,
         writeTimeStampCols.forEach(col -> {
             selectTTLWriteTimeCols.append(",writetime(" + allCols[col] + ")");
         });
-        selectColTypes = getTypes(Util.getSparkProp(sc, KnownProperties.ORIGIN_COLUMN_TYPES));
-        String idCols = Util.getSparkPropOrEmpty(sc, KnownProperties.TARGET_PRIMARY_KEY);
+        selectColTypes = getTypes(propertyHelper.getAsString(KnownProperties.ORIGIN_COLUMN_TYPES));
+        String idCols = propertyHelper.getAsString(KnownProperties.TARGET_PRIMARY_KEY);
         idColTypes = selectColTypes.subList(0, idCols.split(",").length);
 
-        String insertCols = Util.getSparkPropOrEmpty(sc, KnownProperties.TARGET_COLUMN_NAMES);
+        String insertCols = propertyHelper.getAsString(KnownProperties.TARGET_COLUMN_NAMES);
         if (null == insertCols || insertCols.trim().isEmpty()) {
             insertCols = selectCols;
         }
@@ -153,15 +140,15 @@ protected AbstractJobSession(CqlSession originSession, CqlSession targetSession,
                 "select " + insertCols + " from " + targetKeyspaceTable
                         + " where " + insertBinds);
 
-        hasRandomPartitioner = Boolean.parseBoolean(Util.getSparkPropOr(sc, KnownProperties.ORIGIN_HAS_RANDOM_PARTITIONER, "false"));
-        isCounterTable = Boolean.parseBoolean(Util.getSparkPropOr(sc, KnownProperties.ORIGIN_IS_COUNTER, "false"));
+        hasRandomPartitioner = propertyHelper.getBoolean(KnownProperties.ORIGIN_HAS_RANDOM_PARTITIONER);
+        isCounterTable = propertyHelper.getBoolean(KnownProperties.ORIGIN_IS_COUNTER);
         if (isCounterTable) {
-            String updateSelectMappingStr = Util.getSparkPropOr(sc, KnownProperties.ORIGIN_COUNTER_INDEXES, "0");
+            String updateSelectMappingStr = propertyHelper.getString(KnownProperties.ORIGIN_COUNTER_INDEXES);
             for (String updateSelectIndex : updateSelectMappingStr.split(",")) {
                 updateSelectMapping.add(Integer.parseInt(updateSelectIndex));
             }
 
-            String counterTableUpdate = Util.getSparkProp(sc, KnownProperties.ORIGIN_COUNTER_CQL);
+            String counterTableUpdate = propertyHelper.getString(KnownProperties.ORIGIN_COUNTER_CQL);
             targetInsertStatement = targetSession.prepare(counterTableUpdate);
         } else {
             insertBinds = "";
@@ -186,7 +173,7 @@ protected AbstractJobSession(CqlSession originSession, CqlSession targetSession,
         }
 
         // Handle rows with blank values for 'timestamp' data-type in primary-key fields
-        tsReplaceValStr = Util.getSparkPropOr(sc, KnownProperties.TARGET_REPLACE_MISSING_TS, "");
+        tsReplaceValStr = propertyHelper.getAsString(KnownProperties.TARGET_REPLACE_MISSING_TS);
         if (!tsReplaceValStr.isEmpty()) {
             tsReplaceVal = Long.parseLong(tsReplaceValStr);
         }
diff --git a/src/main/java/datastax/astra/migrate/BaseJobSession.java b/src/main/java/datastax/astra/migrate/BaseJobSession.java
@@ -6,6 +6,7 @@
 import com.datastax.oss.driver.api.core.cql.Row;
 import com.datastax.oss.driver.shaded.guava.common.util.concurrent.RateLimiter;
 import datastax.astra.migrate.properties.KnownProperties;
+import datastax.astra.migrate.properties.PropertyHelper;
 import org.apache.spark.SparkConf;
 
 import java.util.ArrayList;
@@ -15,6 +16,8 @@
 
 public abstract class BaseJobSession {
 
+    protected PropertyHelper propertyHelper = PropertyHelper.getInstance();
+
     protected PreparedStatement originSelectStatement;
     protected PreparedStatement targetSelectStatement;
     protected PreparedStatement targetInsertStatement;
@@ -65,8 +68,9 @@ public abstract class BaseJobSession {
     protected long tsReplaceVal;
 
     protected BaseJobSession(SparkConf sc) {
-        readConsistencyLevel = Util.mapToConsistencyLevel(Util.getSparkPropOrEmpty(sc, KnownProperties.READ_CL));
-        writeConsistencyLevel = Util.mapToConsistencyLevel(Util.getSparkPropOrEmpty(sc, KnownProperties.WRITE_CL));
+        propertyHelper.initializeSparkConf(sc);
+        readConsistencyLevel = Util.mapToConsistencyLevel(propertyHelper.getString(KnownProperties.READ_CL));
+        writeConsistencyLevel = Util.mapToConsistencyLevel(propertyHelper.getString(KnownProperties.WRITE_CL));
     }
 
     public String getKey(Row originRow) {
diff --git a/src/main/java/datastax/astra/migrate/CopyJobSession.java b/src/main/java/datastax/astra/migrate/CopyJobSession.java
@@ -24,11 +24,11 @@ public class CopyJobSession extends AbstractJobSession {
 
     protected CopyJobSession(CqlSession originSession, CqlSession targetSession, SparkConf sc) {
         super(originSession, targetSession, sc);
-        filterData = Boolean.parseBoolean(sc.get(KnownProperties.ORIGIN_FILTER_COLUMN_ENABLED, "false"));
-        filterColName = Util.getSparkPropOrEmpty(sc, KnownProperties.ORIGIN_FILTER_COLUMN_NAME);
-        filterColType = Util.getSparkPropOrEmpty(sc, KnownProperties.ORIGIN_FILTER_COLUMN_TYPE);
-        filterColIndex = Integer.parseInt(sc.get(KnownProperties.ORIGIN_FILTER_COLUMN_INDEX, "0"));
-        filterColValue = Util.getSparkPropOrEmpty(sc, KnownProperties.ORIGIN_FILTER_COLUMN_VALUE);
+        filterData = propertyHelper.getBoolean(KnownProperties.ORIGIN_FILTER_COLUMN_ENABLED);
+        filterColName = propertyHelper.getAsString(KnownProperties.ORIGIN_FILTER_COLUMN_NAME);
+        filterColType = propertyHelper.getAsString(KnownProperties.ORIGIN_FILTER_COLUMN_TYPE);
+        filterColIndex = propertyHelper.getInteger(KnownProperties.ORIGIN_FILTER_COLUMN_INDEX);
+        filterColValue = propertyHelper.getAsString(KnownProperties.ORIGIN_FILTER_COLUMN_VALUE);
     }
 
     public static CopyJobSession getInstance(CqlSession originSession, CqlSession targetSession, SparkConf sc) {
diff --git a/src/main/java/datastax/astra/migrate/DiffJobSession.java b/src/main/java/datastax/astra/migrate/DiffJobSession.java
@@ -37,10 +37,10 @@ public class DiffJobSession extends CopyJobSession {
     private DiffJobSession(CqlSession originSession, CqlSession targetSession, SparkConf sc) {
         super(originSession, targetSession, sc);
 
-        autoCorrectMissing = Boolean.parseBoolean(Util.getSparkPropOr(sc, KnownProperties.TARGET_AUTOCORRECT_MISSING, "false"));
+        autoCorrectMissing = propertyHelper.getBoolean(KnownProperties.TARGET_AUTOCORRECT_MISSING);
         logger.info("PARAM -- Autocorrect Missing: {}", autoCorrectMissing);
 
-        autoCorrectMismatch = Boolean.parseBoolean(Util.getSparkPropOr(sc, KnownProperties.TARGET_AUTOCORRECT_MISMATCH, "false"));
+        autoCorrectMismatch = propertyHelper.getBoolean(KnownProperties.TARGET_AUTOCORRECT_MISMATCH);
         logger.info("PARAM -- Autocorrect Mismatch: {}", autoCorrectMismatch);
     }
 
diff --git a/src/main/java/datastax/astra/migrate/OriginCountJobSession.java b/src/main/java/datastax/astra/migrate/OriginCountJobSession.java
@@ -30,35 +30,37 @@ public class OriginCountJobSession extends BaseJobSession {
     protected OriginCountJobSession(CqlSession originSession, SparkConf sc) {
         super(sc);
         this.originSessionSession = originSession;
-        batchSize = new Integer(sc.get(KnownProperties.SPARK_BATCH_SIZE, "1"));
-        printStatsAfter = new Integer(sc.get(KnownProperties.SPARK_STATS_AFTER, "100000"));
-        if (printStatsAfter < 1) {
-            printStatsAfter = 100000;
+        batchSize = propertyHelper.getInteger(KnownProperties.SPARK_BATCH_SIZE);
+        printStatsAfter = propertyHelper.getInteger(KnownProperties.SPARK_STATS_AFTER);
+        if (!propertyHelper.meetsMinimum(KnownProperties.SPARK_STATS_AFTER, printStatsAfter, 1)) {
+            logger.warn(KnownProperties.SPARK_STATS_AFTER +" must be greater than 0.  Setting to default value of " + KnownProperties.getDefaultAsString(KnownProperties.SPARK_STATS_AFTER));
+            propertyHelper.setProperty(KnownProperties.SPARK_STATS_AFTER, KnownProperties.getDefault(KnownProperties.SPARK_STATS_AFTER));
+            printStatsAfter = propertyHelper.getInteger(KnownProperties.SPARK_STATS_AFTER);
         }
 
-        readLimiter = RateLimiter.create(new Integer(sc.get(KnownProperties.SPARK_LIMIT_READ, "20000")));
-        originKeyspaceTable = sc.get(KnownProperties.ORIGIN_KEYSPACE_TABLE);
+        readLimiter = RateLimiter.create(propertyHelper.getInteger(KnownProperties.SPARK_LIMIT_READ));
+        originKeyspaceTable = propertyHelper.getString(KnownProperties.ORIGIN_KEYSPACE_TABLE);
 
-        hasRandomPartitioner = Boolean.parseBoolean(sc.get(KnownProperties.ORIGIN_HAS_RANDOM_PARTITIONER, "false"));
-        isCounterTable = Boolean.parseBoolean(sc.get(KnownProperties.ORIGIN_IS_COUNTER, "false"));
+        hasRandomPartitioner = propertyHelper.getBoolean(KnownProperties.ORIGIN_HAS_RANDOM_PARTITIONER);
+        isCounterTable = propertyHelper.getBoolean(KnownProperties.ORIGIN_IS_COUNTER);
 
-        checkTableforColSize = Boolean.parseBoolean(sc.get(KnownProperties.ORIGIN_CHECK_COLSIZE_ENABLED, "false"));
-        checkTableforselectCols = sc.get(KnownProperties.ORIGIN_CHECK_COLSIZE_COLUMN_NAMES);
-        checkTableforColSizeTypes = getTypes(sc.get(KnownProperties.ORIGIN_CHECK_COLSIZE_COLUMN_TYPES));
-        filterColName = Util.getSparkPropOrEmpty(sc, KnownProperties.ORIGIN_FILTER_COLUMN_NAME);
-        filterColType = Util.getSparkPropOrEmpty(sc, KnownProperties.ORIGIN_FILTER_COLUMN_TYPE);
-        filterColIndex = Integer.parseInt(sc.get(KnownProperties.ORIGIN_FILTER_COLUMN_INDEX, "0"));
-        fieldGuardraillimitMB = Integer.parseInt(sc.get(KnownProperties.FIELD_GUARDRAIL_MB, "0"));
+        checkTableforColSize = propertyHelper.getBoolean(KnownProperties.ORIGIN_CHECK_COLSIZE_ENABLED);
+        checkTableforselectCols = propertyHelper.getAsString(KnownProperties.ORIGIN_CHECK_COLSIZE_COLUMN_NAMES);
+        checkTableforColSizeTypes = getTypes(propertyHelper.getAsString(KnownProperties.ORIGIN_CHECK_COLSIZE_COLUMN_TYPES));
+        filterColName = propertyHelper.getAsString(KnownProperties.ORIGIN_FILTER_COLUMN_NAME);
+        filterColType = propertyHelper.getAsString(KnownProperties.ORIGIN_FILTER_COLUMN_TYPE);
+        filterColIndex = propertyHelper.getInteger(KnownProperties.ORIGIN_FILTER_COLUMN_INDEX);
+        fieldGuardraillimitMB = propertyHelper.getInteger(KnownProperties.FIELD_GUARDRAIL_MB);
 
-        String partionKey = sc.get(KnownProperties.ORIGIN_PARTITION_KEY);
-        idColTypes = getTypes(sc.get(KnownProperties.TARGET_PRIMARY_KEY_TYPES));
+        String partionKey = propertyHelper.getAsString(KnownProperties.ORIGIN_PARTITION_KEY);
+        idColTypes = getTypes(propertyHelper.getAsString(KnownProperties.TARGET_PRIMARY_KEY_TYPES));
 
-        String selectCols = sc.get(KnownProperties.ORIGIN_COLUMN_NAMES);
-        String updateSelectMappingStr = sc.get(KnownProperties.ORIGIN_COUNTER_INDEXES, "0");
+        String selectCols = propertyHelper.getAsString(KnownProperties.ORIGIN_COLUMN_NAMES);
+        String updateSelectMappingStr = propertyHelper.getAsString(KnownProperties.ORIGIN_COUNTER_INDEXES);
         for (String updateSelectIndex : updateSelectMappingStr.split(",")) {
             updateSelectMapping.add(Integer.parseInt(updateSelectIndex));
         }
-        String originSelectCondition = sc.get(KnownProperties.ORIGIN_FILTER_CONDITION, "");
+        String originSelectCondition = propertyHelper.getAsString(KnownProperties.ORIGIN_FILTER_CONDITION);
         // TODO: AbstractJobSession has some checks to ensure AND is added to the condition
         originSelectStatement = originSession.prepare(
                 "select " + selectCols + " from " + originKeyspaceTable + " where token(" + partionKey.trim()
diff --git a/src/main/java/datastax/astra/migrate/Util.java b/src/main/java/datastax/astra/migrate/Util.java
@@ -1,38 +1,14 @@
 package datastax.astra.migrate;
 
 import com.datastax.oss.driver.api.core.ConsistencyLevel;
-import datastax.astra.migrate.properties.KnownProperties;
-import datastax.astra.migrate.properties.PropertyHelper;
 import org.apache.commons.lang.StringUtils;
-import org.apache.spark.SparkConf;
 
 import java.io.BufferedReader;
 import java.io.FileNotFoundException;
 import java.io.FileReader;
-import java.util.NoSuchElementException;
 
 public class Util {
 
-    public static String getSparkProp(SparkConf sc, String prop) {
-        String retVal = PropertyHelper.getInstance(sc).getAsString(prop);
-        if (StringUtils.isEmpty(retVal) && (prop.contains("origin") || prop.contains("target"))) {
-            retVal = PropertyHelper.getInstance(sc).getAsString(prop.replace("origin", "source").replace("target", "destination"));
-        }
-        if (!KnownProperties.isKnown(prop)) {
-            throw new IllegalArgumentException("Unknown property: " + prop + "; this is a bug in the code: the property is not configured in KnownProperties.java");
-        }
-        return retVal;
-    }
-
-    public static String getSparkPropOr(SparkConf sc, String prop, String defaultVal) {
-        String retVal = getSparkProp(sc,prop);
-        return StringUtils.isEmpty(retVal) ? defaultVal : retVal;
-    }
-
-    public static String getSparkPropOrEmpty(SparkConf sc, String prop) {
-        return getSparkPropOr(sc, prop, "");
-    }
-
     public static BufferedReader getfileReader(String fileName) {
         try {
             return new BufferedReader(new FileReader(fileName));
diff --git a/src/main/java/datastax/astra/migrate/properties/KnownProperties.java b/src/main/java/datastax/astra/migrate/properties/KnownProperties.java
@@ -196,7 +196,7 @@ public enum PropertyType {
            types.put(SPARK_STATS_AFTER, PropertyType.NUMBER);
         defaults.put(SPARK_STATS_AFTER, "100000");
            types.put(FIELD_GUARDRAIL_MB, PropertyType.NUMBER);
-        defaults.put(FIELD_GUARDRAIL_MB, "10");
+        defaults.put(FIELD_GUARDRAIL_MB, "0");
            types.put(PARTITION_MIN, PropertyType.NUMBER);
         defaults.put(PARTITION_MIN, "-9223372036854775808");
            types.put(PARTITION_MAX, PropertyType.NUMBER);
diff --git a/src/main/scala/datastax/astra/migrate/BaseJob.scala b/src/main/scala/datastax/astra/migrate/BaseJob.scala
