PR review fixes

Author: pravinbhat

src/main/java/datastax/astra/migrate/AbstractJobSession.java

@@ -30,8 +30,8 @@ public class AbstractJobSession extends BaseJobSession {
     protected List<String> ttlWTCols;
     protected String tsReplaceValStr;
     protected long tsReplaceVal;
-    protected Long customWritetime = 0l;
-    protected Long incrWritetime = 0l;
+    protected long customWriteTime = 0l;
+    protected long incrementWriteTime = 0l;
 
     protected AbstractJobSession(CqlSession sourceSession, CqlSession astraSession, SparkConf sc) {
         this(sourceSession, astraSession, sc, false);
@@ -98,19 +98,19 @@ protected AbstractJobSession(CqlSession sourceSession, CqlSession astraSession,
         String customWriteTimeStr =
                 Util.getSparkPropOr(sc, "spark.target.writeTime.fixedValue", "0");
         if (null != customWriteTimeStr && customWriteTimeStr.trim().length() > 0 && StringUtils.isNumeric(customWriteTimeStr.trim())) {
-            customWritetime = Long.parseLong(customWriteTimeStr);
+            customWriteTime = Long.parseLong(customWriteTimeStr);
         }
 
         String incrWriteTimeStr =
                 Util.getSparkPropOr(sc, "spark.target.writeTime.incrementBy", "0");
         if (null != incrWriteTimeStr && incrWriteTimeStr.trim().length() > 0 && StringUtils.isNumeric(incrWriteTimeStr.trim())) {
-            incrWritetime = Long.parseLong(incrWriteTimeStr);
+            incrementWriteTime = Long.parseLong(incrWriteTimeStr);
         }
 
         logger.info("PARAM -- TTL-WriteTime Columns: {}", ttlWTCols);
         logger.info("PARAM -- WriteTimes Filter: {}", writeTimeStampFilter);
-        logger.info("PARAM -- WriteTime Custom Value: {}", customWritetime);
-        logger.info("PARAM -- WriteTime increment Value: {}", incrWritetime);
+        logger.info("PARAM -- WriteTime Custom Value: {}", customWriteTime);
+        logger.info("PARAM -- WriteTime Increment Value: {}", incrementWriteTime);
         if (writeTimeStampFilter) {
             logger.info("PARAM -- minWriteTimeStampFilter: {} datetime is {}", minWriteTimeStampFilter,
                     Instant.ofEpochMilli(minWriteTimeStampFilter / 1000));
@@ -202,10 +202,10 @@ public BoundStatement bindInsert(PreparedStatement insertStatement, Row sourceRo
             if (!ttlWTCols.isEmpty()) {
                 boundInsertStatement = boundInsertStatement.set(index, getLargestTTL(sourceRow), Integer.class);
                 index++;
-                if (customWritetime > 0) {
-                    boundInsertStatement = boundInsertStatement.set(index, customWritetime, Long.class);
+                if (customWriteTime > 0) {
+                    boundInsertStatement = boundInsertStatement.set(index, customWriteTime, Long.class);
                 } else {
-                    boundInsertStatement = boundInsertStatement.set(index, getLargestWriteTimeStamp(sourceRow) + incrWritetime, Long.class);
+                    boundInsertStatement = boundInsertStatement.set(index, getLargestWriteTimeStamp(sourceRow) + incrementWriteTime, Long.class);
                 }
             }
         }

src/main/java/datastax/astra/migrate/BaseJobSession.java

@@ -31,26 +31,26 @@ public abstract class BaseJobSession {
     // Rate = Total Throughput (write/read per sec) / Total Executors
     protected RateLimiter readLimiter;
     protected RateLimiter writeLimiter;
-    protected Integer maxRetries = 10;
+    protected int maxRetries = 10;
     protected AtomicLong readCounter = new AtomicLong(0);
-    protected Integer batchSize = 1;
-    protected Integer fetchSizeInRows = 1000;
-    protected Integer printStatsAfter = 100000;
+    protected int batchSize = 1;
+    protected int fetchSizeInRows = 1000;
+    protected int printStatsAfter;
 
-    protected Boolean writeTimeStampFilter = Boolean.FALSE;
-    protected Long minWriteTimeStampFilter = 0l;
-    protected Long maxWriteTimeStampFilter = Long.MAX_VALUE;
+    protected boolean writeTimeStampFilter;
+    protected long minWriteTimeStampFilter = 0l;
+    protected long maxWriteTimeStampFilter = Long.MAX_VALUE;
 
-    protected Boolean isCounterTable = false;
+    protected boolean isCounterTable;
 
     protected String sourceKeyspaceTable;
     protected String astraKeyspaceTable;
 
-    protected Boolean hasRandomPartitioner;
-    protected Boolean filterData;
+    protected boolean hasRandomPartitioner;
+    protected boolean filterData;
     protected String filterColName;
     protected String filterColType;
-    protected Integer filterColIndex;
+    protected int filterColIndex;
     protected String filterColValue;
     protected String sourceSelectCondition;
 

src/main/java/datastax/astra/migrate/DiffJobSession.java

@@ -30,8 +30,8 @@ public class DiffJobSession extends CopyJobSession {
     private final AtomicLong validCounter = new AtomicLong(0);
     private final AtomicLong skippedCounter = new AtomicLong(0);
     public Logger logger = LoggerFactory.getLogger(this.getClass().getName());
-    protected Boolean autoCorrectMissing = false;
-    protected Boolean autoCorrectMismatch = false;
+    protected boolean autoCorrectMissing;
+    protected boolean autoCorrectMismatch;
 
     private DiffJobSession(CqlSession sourceSession, CqlSession astraSession, SparkConf sc) {
         super(sourceSession, astraSession, sc);

src/main/java/datastax/astra/migrate/GuardrailJobSession.java

@@ -19,7 +19,7 @@ public class GuardrailJobSession extends BaseJobSession {
     protected AtomicLong readCounter = new AtomicLong(0);
     protected AtomicLong largeRowCounter = new AtomicLong(0);
     protected AtomicLong largeFieldCounter = new AtomicLong(0);
-    protected Integer guardrailColSizeInKB;
+    protected int guardrailColSizeInKB;
 
     protected GuardrailJobSession(CqlSession session, SparkConf sc) {
         super(sc);