CDM-67 renaming variable for increased clarity per review comment

Author: mieslep

src/main/java/com/datastax/cdm/job/AbstractJobSession.java

@@ -42,13 +42,13 @@ protected AbstractJobSession(CqlSession originSession, CqlSession targetSession,
             printStatsAfter = propertyHelper.getInteger(KnownProperties.PRINT_STATS_AFTER);
         }
 
-        originLimiter = RateLimiter.create(propertyHelper.getInteger(KnownProperties.PERF_RATELIMIT_ORIGIN));
-        targetLimiter = RateLimiter.create(propertyHelper.getInteger(KnownProperties.PERF_RATELIMIT_TARGET));
+        rateLimiterOrigin = RateLimiter.create(propertyHelper.getInteger(KnownProperties.PERF_RATELIMIT_ORIGIN));
+        rateLimiterTarget = RateLimiter.create(propertyHelper.getInteger(KnownProperties.PERF_RATELIMIT_TARGET));
         maxRetries = propertyHelper.getInteger(KnownProperties.MAX_RETRIES);
 
         logger.info("PARAM -- Max Retries: {}", maxRetries);
-        logger.info("PARAM -- Origin Rate Limit: {}", originLimiter.getRate());
-        logger.info("PARAM -- Target Rate Limit: {}", targetLimiter.getRate());
+        logger.info("PARAM -- Origin Rate Limit: {}", rateLimiterOrigin.getRate());
+        logger.info("PARAM -- Target Rate Limit: {}", rateLimiterTarget.getRate());
 
         this.originSession = new EnhancedSession(propertyHelper, originSession, true);
         this.targetSession = new EnhancedSession(propertyHelper, targetSession, false);

src/main/java/com/datastax/cdm/job/BaseJobSession.java

@@ -24,8 +24,8 @@ public abstract class BaseJobSession {
     // then do the following to set the values as they are only applicable per JVM
     // (hence spark Executor)...
     // Rate = Total Throughput (write/read per sec) / Total Executors
-    protected RateLimiter originLimiter;
-    protected RateLimiter targetLimiter;
+    protected RateLimiter rateLimiterOrigin;
+    protected RateLimiter rateLimiterTarget;
     protected Integer maxRetries = 10;
 
     protected Integer printStatsAfter = 100000;

src/main/java/com/datastax/cdm/job/CopyJobSession.java

@@ -73,7 +73,7 @@ public void getDataAndInsert(BigInteger min, BigInteger max) {
                 Collection<CompletionStage<AsyncResultSet>> writeResults = new ArrayList<>();
 
                 for (Row originRow : resultSet) {
-                    originLimiter.acquire(1);
+                    rateLimiterOrigin.acquire(1);
                     readCnt++;
                     if (readCnt % printStatsAfter == 0) {
                         printCounts(false);
@@ -101,7 +101,7 @@ public void getDataAndInsert(BigInteger min, BigInteger max) {
                             continue;
                         }
 
-                        targetLimiter.acquire(1);
+                        rateLimiterTarget.acquire(1);
                         batch = writeAsync(batch, writeResults, boundUpsert);
                         unflushedWrites++;
 
@@ -165,7 +165,7 @@ private void flushAndClearWrites(BatchStatement batch, Collection<CompletionStag
 
     private BoundStatement bind(Record r) {
         if (isCounterTable) {
-            targetLimiter.acquire(1);
+            rateLimiterTarget.acquire(1);
             Record targetRecord = targetSelectByPKStatement.getRecord(r.getPk());
             if (null != targetRecord) {
                 r.setTargetRow(targetRecord.getTargetRow());

src/main/java/com/datastax/cdm/job/CopyPKJobSession.java

@@ -56,7 +56,7 @@ public void getRowAndInsert(SplitPartitions.PKRows rowsList) {
                 return;
             }
 
-            originLimiter.acquire(1);
+            rateLimiterOrigin.acquire(1);
             Record recordFromOrigin = originSelectByPKStatement.getRecord(pk);
             if (null == recordFromOrigin) {
                 missingCounter.incrementAndGet();
@@ -80,7 +80,7 @@ public void getRowAndInsert(SplitPartitions.PKRows rowsList) {
                 }
             }
 
-            targetLimiter.acquire(1);
+            rateLimiterTarget.acquire(1);
             targetSession.getTargetUpsertStatement().putRecord(record);
             writeCounter.incrementAndGet();
 

src/main/java/com/datastax/cdm/job/DiffJobSession.java

@@ -112,7 +112,7 @@ public void getDataAndDiff(BigInteger min, BigInteger max) {
 
                 List<Record> recordsToDiff = new ArrayList<>(fetchSizeInRows);
                 StreamSupport.stream(resultSet.spliterator(), false).forEach(originRow -> {
-                    originLimiter.acquire(1);
+                    rateLimiterOrigin.acquire(1);
                     Record record = new Record(pkFactory.getTargetPK(originRow), originRow, null);
 
                     if (originSelectByPartitionRangeStatement.shouldFilterRecord(record)) {
@@ -132,7 +132,7 @@ public void getDataAndDiff(BigInteger min, BigInteger max) {
                                 }
                             }
 
-                            targetLimiter.acquire(1);
+                            rateLimiterTarget.acquire(1);
                             CompletionStage<AsyncResultSet> targetResult = targetSelectByPKStatement.getAsyncResult(r.getPk());
 
                             if (null==targetResult) {
@@ -203,7 +203,7 @@ private void diff(Record record) {
 
             //correct data
             if (autoCorrectMissing) {
-                targetLimiter.acquire(1);
+                rateLimiterTarget.acquire(1);
                 targetSession.getTargetUpsertStatement().putRecord(record);
                 correctedMissingCounter.incrementAndGet();
                 logger.error("Inserted missing row in target: {}", record.getPk());
@@ -217,7 +217,7 @@ private void diff(Record record) {
             logger.error("Mismatch row found for key: {} Mismatch: {}", record.getPk(), diffData);
 
             if (autoCorrectMismatch) {
-                targetLimiter.acquire(1);
+                rateLimiterTarget.acquire(1);
                 targetSession.getTargetUpsertStatement().putRecord(record);
                 correctedMismatchCounter.incrementAndGet();
                 logger.error("Corrected mismatch row in target: {}", record.getPk());

src/main/java/com/datastax/cdm/job/GuardrailCheckJobSession.java

@@ -50,7 +50,7 @@ public void guardrailCheck(BigInteger min, BigInteger max) {
             ResultSet resultSet = originSelectByPartitionRangeStatement.execute(originSelectByPartitionRangeStatement.bind(min, max));
             String checkString;
             for (Row originRow : resultSet) {
-                originLimiter.acquire(1);
+                rateLimiterOrigin.acquire(1);
                 readCounter.addAndGet(1);
 
                 if (readCounter.get() % printStatsAfter == 0) {