CDM-67 adding target read rate limiter

Author: mieslep

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

@@ -44,11 +44,15 @@ protected AbstractJobSession(CqlSession originSession, CqlSession targetSession,
 
         readLimiter = RateLimiter.create(propertyHelper.getInteger(KnownProperties.PERF_LIMIT_READ));
         writeLimiter = RateLimiter.create(propertyHelper.getInteger(KnownProperties.PERF_LIMIT_WRITE));
+        Integer readLimitTarget = propertyHelper.getInteger(KnownProperties.PERF_LIMIT_READ_TARGET);
+        if (readLimitTarget == null || readLimitTarget < 0) { readLimitTarget = propertyHelper.getInteger(KnownProperties.PERF_LIMIT_READ);}
+        readLimiterTarget = RateLimiter.create(readLimitTarget);
         maxRetries = propertyHelper.getInteger(KnownProperties.MAX_RETRIES);
 
         logger.info("PARAM -- Max Retries: {}", maxRetries);
         logger.info("PARAM -- ReadRateLimit: {}", readLimiter.getRate());
         logger.info("PARAM -- WriteRateLimit: {}", writeLimiter.getRate());
+        logger.info("PARAM -- TargetReadRateLimit: {}", readLimiterTarget.getRate());
 
         this.originSession = new EnhancedSession(propertyHelper, originSession, true);
         this.targetSession = new EnhancedSession(propertyHelper, targetSession, false);

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

@@ -25,6 +25,7 @@ public abstract class BaseJobSession {
     // (hence spark Executor)...
     // Rate = Total Throughput (write/read per sec) / Total Executors
     protected RateLimiter readLimiter;
+    protected RateLimiter readLimiterTarget;
     protected RateLimiter writeLimiter;
     protected Integer maxRetries = 10;
 

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

@@ -169,6 +169,7 @@ private void flushAndClearWrites(Collection<CompletionStage<AsyncResultSet>> wri
 
     private BoundStatement bind(Record r) {
         if (isCounterTable) {
+            readLimiterTarget.acquire(1);
             Record targetRecord = targetSelectByPKStatement.getRecord(r.getPk());
             if (null != targetRecord) {
                 r.setTargetRow(targetRecord.getTargetRow());

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

@@ -132,6 +132,7 @@ public void getDataAndDiff(BigInteger min, BigInteger max) {
                                 }
                             }
 
+                            readLimiterTarget.acquire(1);
                             CompletionStage<AsyncResultSet> targetResult = targetSelectByPKStatement.getAsyncResult(r.getPk());
 
                             if (null==targetResult) {

src/main/java/com/datastax/cdm/properties/KnownProperties.java

@@ -100,6 +100,7 @@ public enum PropertyType {
     public static final String PERF_NUM_PARTS                   = "spark.cdm.perfops.numParts";             // 10000, was spark.splitSize
     public static final String PERF_BATCH_SIZE                  = "spark.cdm.perfops.batchSize";             // 5
     public static final String PERF_LIMIT_READ                  = "spark.cdm.perfops.readRateLimit";         // 20000
+    public static final String PERF_LIMIT_READ_TARGET           = "spark.cdm.perfops.readRateLimit.target";  // readRateLimit
     public static final String PERF_LIMIT_WRITE                 = "spark.cdm.perfops.writeRateLimit";        // 40000
 
     public static final String READ_CL                          = "spark.cdm.perfops.consistency.read";
@@ -122,6 +123,7 @@ public enum PropertyType {
         defaults.put(PERF_BATCH_SIZE, "5");
            types.put(PERF_LIMIT_READ, PropertyType.NUMBER);
         defaults.put(PERF_LIMIT_READ, "20000");
+           types.put(PERF_LIMIT_READ_TARGET, PropertyType.NUMBER);
            types.put(PERF_LIMIT_WRITE, PropertyType.NUMBER);
         defaults.put(PERF_LIMIT_WRITE, "40000");
 

src/resources/sparkConf.properties

@@ -148,8 +148,11 @@ spark.cdm.autocorrect.mismatch                    false
 #                           If .batchSize would mean that more than 1 partition is often contained in a batch,  
 #                           the figure should be reduced. Ideally < 1% of batches have more than 1 partition.
 #    .readRateLimit       : Defaults to 20000. Concurrent number of records that may will be read across 
-#                           all parallel threads. This may be adjusted up (or down), depending on the amount 
-#                           of data and the processing capacity of the Origin cluster.
+#                           all parallel threads from Origin. This may be adjusted up (or down), depending on
+#                           the amount of data and the processing capacity of the Origin cluster.
+#    .readRateLimitTarget : Defaults to readRateLimit. Concurrent number of records that may will be read across
+#                           all parallel threads from Target. This may be adjusted up (or down), depending on
+#                           the amount of data and the processing capacity of the Target cluster.
 #    .writeRateLimit      : Defaults to 40000. Concurrent number of records that may will be written across
 #                           all parallel threads. This may be adjusted up (or down), depending on the amount 
 #                           of data and the processing capacity of the Target cluster.
@@ -173,6 +176,7 @@ spark.cdm.autocorrect.mismatch                    false
 spark.cdm.perfops.numParts                        10000
 spark.cdm.perfops.batchSize                       5
 spark.cdm.perfops.readRateLimit                   20000
+#spark.cdm.perfops.readRateLimitTarget             20000
 spark.cdm.perfops.writeRateLimit                  40000
 #spark.cdm.perfops.consistency.read                LOCAL_QUORUM
 #spark.cdm.perfops.consistency.write               LOCAL_QUORUM