W-17312191: Set kinesis.consumer.traceback.minutes to 20

Author: peterzxu-crm

carbonj.service/src/main/java/com/demandware/carbonj/service/db/points/DataPointArchiveRocksDB.java

@@ -324,9 +324,6 @@ public int put( DataPoints points )
                     int interval = policy.interval(p.ts);
                     byte[] key = DataPointRecord.toKeyBytes(p.metricId, interval, longId);
                     byte[] value = DataPointRecord.toValueBytes(p.val);
-                    if (p.name.startsWith("pod262.ecom.bfdk.bfdk_prd")) {
-                        log.info("=====>>>>> {} {} {} {}", p.name, p.val, p.ts, p.metricId);
-                    }
                     batch.put(key, value);
                     String namespace = nameUtils.firstSegment(p.name);
                     if (!latencyByNamespaceMap.containsKey(namespace)) {

carbonj.service/src/main/java/com/demandware/carbonj/service/engine/Consumers.java

@@ -52,11 +52,11 @@ public class Consumers {
 
     private final String carbonjEnv;
 
-    private final int kinesisConsumerTracebackSeconds;
+    private final int kinesisConsumerTracebackMinutes;
 
     Consumers(MetricRegistry metricRegistry, PointProcessor pointProcessor, PointProcessor recoveryPointProcessor, File rulesFile,
               KinesisConfig kinesisConfig, CheckPointMgr<Date> checkPointMgr, String kinesisConsumerRegion,
-              NamespaceCounter namespaceCounter, File indexNameSyncDir, String carbonjEnv, int kinesisConsumerTracebackSeconds) {
+              NamespaceCounter namespaceCounter, File indexNameSyncDir, String carbonjEnv, int kinesisConsumerTracebackMinutes) {
 
         this.metricRegistry = metricRegistry;
         this.pointProcessor = pointProcessor;
@@ -69,7 +69,7 @@ public class Consumers {
         this.consumers = new ConcurrentHashMap<>();
         this.consumerRules = new ConsumerRules(rulesFile);
         this.carbonjEnv = carbonjEnv;
-        this.kinesisConsumerTracebackSeconds = kinesisConsumerTracebackSeconds;
+        this.kinesisConsumerTracebackMinutes = kinesisConsumerTracebackMinutes;
         reload();
     }
 
@@ -147,7 +147,7 @@ private void reconfigureConsumers(Set<String> newRules, Set<String> currentRules
 
                 Counter initRetryCounter = metricRegistry.counter(MetricRegistry.name("kinesis.consumer." + kinesisStreamName + ".initRetryCounter"));
                 KinesisConsumer kinesisConsumer = new KinesisConsumer(metricRegistry, pointProcessor, recoveryPointProcessor, kinesisStreamName,
-                        kinesisApplicationName, kinesisConfig, checkPointMgr, initRetryCounter, kinesisConsumerRegion, kinesisConsumerTracebackSeconds);
+                        kinesisApplicationName, kinesisConfig, checkPointMgr, initRetryCounter, kinesisConsumerRegion, kinesisConsumerTracebackMinutes);
                 log.info("New Consumer created with name {}", kinesisStreamName);
                 newConsumers.add(consumerName);
                 consumers.put(consumerName, kinesisConsumer);

carbonj.service/src/main/java/com/demandware/carbonj/service/engine/KinesisConsumer.java

@@ -40,6 +40,8 @@
 
 import java.net.URI;
 import java.time.Clock;
+import java.time.Duration;
+import java.time.Instant;
 import java.util.Date;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
@@ -70,21 +72,21 @@ public class KinesisConsumer extends Thread {
 
     private final String overrideKinesisEndpoint;
 
-    private final int kinesisConsumerTracebackSeconds;
+    private final int kinesisConsumerTracebackMinutes;
 
     public KinesisConsumer(MetricRegistry metricRegistry, PointProcessor pointProcessor, PointProcessor recoveryPointProcessor,
                            String kinesisStreamName, String kinesisApplicationName,
                            KinesisConfig kinesisConfig, CheckPointMgr<Date> checkPointMgr,
-                           Counter noOfRestarts, String kinesisConsumerRegion, int kinesisConsumerTracebackSeconds) {
+                           Counter noOfRestarts, String kinesisConsumerRegion, int kinesisConsumerTracebackMinutes) {
         this(metricRegistry, pointProcessor, recoveryPointProcessor, kinesisStreamName, kinesisApplicationName, kinesisConfig,
-                checkPointMgr, noOfRestarts, kinesisConsumerRegion, kinesisConsumerTracebackSeconds, null);
+                checkPointMgr, noOfRestarts, kinesisConsumerRegion, kinesisConsumerTracebackMinutes, null);
     }
 
     public KinesisConsumer(MetricRegistry metricRegistry, PointProcessor pointProcessor, PointProcessor recoveryPointProcessor,
                            String kinesisStreamName, String kinesisApplicationName,
                            KinesisConfig kinesisConfig, CheckPointMgr<Date> checkPointMgr,
                            Counter noOfRestarts, String kinesisConsumerRegion,
-                           int kinesisConsumerTracebackSeconds,
+                           int kinesisConsumerTracebackMinutes,
                            String overrideKinesisEndpoint) {
         this.metricRegistry = metricRegistry;
         this.pointProcessor = Preconditions.checkNotNull(pointProcessor);
@@ -95,7 +97,7 @@ public KinesisConsumer(MetricRegistry metricRegistry, PointProcessor pointProces
         this.checkPointMgr = checkPointMgr;
         this.noOfRestarts = noOfRestarts;
         this.kinesisConsumerRegion = kinesisConsumerRegion;
-        this.kinesisConsumerTracebackSeconds = kinesisConsumerTracebackSeconds;
+        this.kinesisConsumerTracebackMinutes = kinesisConsumerTracebackMinutes;
         this.overrideKinesisEndpoint = overrideKinesisEndpoint;
         log.info("Kinesis consumer started");
         this.start();
@@ -160,9 +162,9 @@ public void run () {
                 LeaseManagementConfig leaseManagementConfig = configsBuilder.leaseManagementConfig()
                         .failoverTimeMillis(kinesisConfig.getLeaseExpirationTimeInSecs() * 1000L);
                 // Since v2 will create a new DynamoDB table, we need to trace back the initial position
+                Instant startTime = Instant.now().minus(Duration.ofMinutes(kinesisConsumerTracebackMinutes));
                 InitialPositionInStreamExtended initialPositionInStreamExtended =
-                        InitialPositionInStreamExtended.newInitialPositionAtTimestamp(
-                                new Date(Clock.systemUTC().millis() - kinesisConsumerTracebackSeconds * 1000L));
+                        InitialPositionInStreamExtended.newInitialPositionAtTimestamp(new Date(startTime.toEpochMilli()));
                 worker = new Scheduler(
                         configsBuilder.checkpointConfig(),
                         configsBuilder.coordinatorConfig(),

carbonj.service/src/main/java/com/demandware/carbonj/service/engine/cfgCarbonJ.java

@@ -210,8 +210,8 @@ public class cfgCarbonJ
     @Value("${carbonj.env:dev}")
     private String carbonjEnv;
 
-    @Value("${kinesis.consumer.traceback.seconds:3600}")
-    private int kinesisConsumerTracebackSeconds;
+    @Value("${kinesis.consumer.traceback.minutes:20}")
+    private int kinesisConsumerTracebackMinutes;
 
     @Bean
     public RestTemplate restTemplate() {
@@ -430,7 +430,7 @@ Consumers consumer( PointProcessor pointProcessor,
             File rulesFile = locateConfigFile( serviceDir, consumerRulesFile );
             Consumers consumer = new Consumers( metricRegistry, pointProcessor, recoveryPointProcessor, rulesFile,
                     kinesisConfig, checkPointMgr, kinesisConsumerRegion,
-                    nsCounter, dataDir == null ? null : new File(dataDir, "index-name-sync"), carbonjEnv, kinesisConsumerTracebackSeconds);
+                    nsCounter, dataDir == null ? null : new File(dataDir, "index-name-sync"), carbonjEnv, kinesisConsumerTracebackMinutes);
             s.scheduleWithFixedDelay( consumer::reload, 15, 30, TimeUnit.SECONDS );
             if (syncSecondaryDb) {
                 s.scheduleWithFixedDelay( consumer::syncNamespaces, 60, 60, TimeUnit.SECONDS );