W-17312191: Set initialPositionInStreamExtended back 1 hour

peterzxu-crm · peterzxu-crm · commit 086f03002996 · 2025-09-05T16:44:55.000-04:00
diff --git a/carbonj.service/src/main/java/com/demandware/carbonj/service/engine/Consumers.java b/carbonj.service/src/main/java/com/demandware/carbonj/service/engine/Consumers.java
@@ -52,9 +52,11 @@ public class Consumers {
 
     private final String carbonjEnv;
 
+    private final int kinesisConsumerTracebackSeconds;
+
     Consumers(MetricRegistry metricRegistry, PointProcessor pointProcessor, PointProcessor recoveryPointProcessor, File rulesFile,
               KinesisConfig kinesisConfig, CheckPointMgr<Date> checkPointMgr, String kinesisConsumerRegion,
-              NamespaceCounter namespaceCounter, File indexNameSyncDir, String carbonjEnv) {
+              NamespaceCounter namespaceCounter, File indexNameSyncDir, String carbonjEnv, int kinesisConsumerTracebackSeconds) {
 
         this.metricRegistry = metricRegistry;
         this.pointProcessor = pointProcessor;
@@ -67,6 +69,7 @@ public class Consumers {
         this.consumers = new ConcurrentHashMap<>();
         this.consumerRules = new ConsumerRules(rulesFile);
         this.carbonjEnv = carbonjEnv;
+        this.kinesisConsumerTracebackSeconds = kinesisConsumerTracebackSeconds;
         reload();
     }
 
@@ -84,7 +87,7 @@ public synchronized void reload() {
                 log.debug(" Check for consumer configuration update");
             }
             if (!currentRules.equals(newRules)) {
-                log.info(String.format("Updating consumer Rules. Old consumer rules: [%s], New consumer rules: [%s]", currentRules, newRules));
+                log.info("Updating consumer Rules. Old consumer rules: [{}], New consumer rules: [{}]", currentRules, newRules);
             }
             else {
                 log.debug(" Consumer rules haven't changed.");
@@ -103,11 +106,11 @@ private void reconfigureConsumers(Set<String> newRules, Set<String> currentRules
             log.info (consumer);
 
             if (newRules.contains(consumer)) {
-                log.info(String.format("[%s] Reuse unchanged consumer", consumer));
+                log.info("[{}] Reuse unchanged consumer", consumer);
                 newConsumers.add(consumer);
                 newRules.remove(consumer);
             } else {
-                log.info(String.format("[%s] Consumer scheduled for removal", consumer));
+                log.info("[{}] Consumer scheduled for removal", consumer);
                 obsoleteConsumers.add(consumer);
             }
 
@@ -117,7 +120,7 @@ private void reconfigureConsumers(Set<String> newRules, Set<String> currentRules
         // we use the host name to generate the kinesis application name as they are stable for stable set pods.
         String hostName = getHostName();
         for (String consumerName : newRules) {
-            log.info(String.format("Creating new consumer with kinesis stream name: %s", consumerName));
+            log.info("Creating new consumer with kinesis stream name: {}", consumerName);
 
             if (consumerName.startsWith("kinesis:")) {
 
@@ -128,23 +131,23 @@ private void reconfigureConsumers(Set<String> newRules, Set<String> currentRules
 
                 try {
                     InputStream input = new FileInputStream(consumerCfgFile);
-                    log.info(" Loading values from " + consumerCfgFile);
+                    log.info(" Loading values from {}", consumerCfgFile);
                     Properties consumerCfg = new Properties();
                     consumerCfg.load(input);
                     String kinesisApplicationNamePropValue = consumerCfg.getProperty("kinesis.application.name");
                     if( kinesisApplicationNamePropValue != null ) {
                         kinesisApplicationName = kinesisApplicationNamePropValue;
                     }
                 } catch (FileNotFoundException e) {
-                    log.warn(consumerCfgFile + " not found in the classpath ");
+                    log.warn("{} not found in the classpath ", consumerCfgFile);
                     log.info(" Falling back to default values ");
                 } catch (Throwable e) {
                     log.error(e.getMessage(), e);
                 }
 
                 Counter initRetryCounter = metricRegistry.counter(MetricRegistry.name("kinesis.consumer." + kinesisStreamName + ".initRetryCounter"));
                 KinesisConsumer kinesisConsumer = new KinesisConsumer(metricRegistry, pointProcessor, recoveryPointProcessor, kinesisStreamName,
-                        kinesisApplicationName, kinesisConfig, checkPointMgr, initRetryCounter, kinesisConsumerRegion);
+                        kinesisApplicationName, kinesisConfig, checkPointMgr, initRetryCounter, kinesisConsumerRegion, kinesisConsumerTracebackSeconds);
                 log.info("New Consumer created with name {}", kinesisStreamName);
                 newConsumers.add(consumerName);
                 consumers.put(consumerName, kinesisConsumer);
diff --git a/carbonj.service/src/main/java/com/demandware/carbonj/service/engine/KinesisConsumer.java b/carbonj.service/src/main/java/com/demandware/carbonj/service/engine/KinesisConsumer.java
@@ -28,7 +28,6 @@
 import software.amazon.kinesis.coordinator.Scheduler;
 import software.amazon.kinesis.leases.LeaseManagementConfig;
 import software.amazon.kinesis.processor.ShardRecordProcessorFactory;
-import software.amazon.kinesis.common.InitialPositionInStream;
 import software.amazon.kinesis.common.InitialPositionInStreamExtended;
 import software.amazon.kinesis.retrieval.polling.PollingConfig;
 import com.codahale.metrics.Counter;
@@ -40,6 +39,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.net.URI;
+import java.time.Clock;
 import java.util.Date;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
@@ -70,18 +70,21 @@ public class KinesisConsumer extends Thread {
 
     private final String overrideKinesisEndpoint;
 
+    private final int kinesisConsumerTracebackSeconds;
+
     public KinesisConsumer(MetricRegistry metricRegistry, PointProcessor pointProcessor, PointProcessor recoveryPointProcessor,
                            String kinesisStreamName, String kinesisApplicationName,
                            KinesisConfig kinesisConfig, CheckPointMgr<Date> checkPointMgr,
-                           Counter noOfRestarts, String kinesisConsumerRegion) {
+                           Counter noOfRestarts, String kinesisConsumerRegion, int kinesisConsumerTracebackSeconds) {
         this(metricRegistry, pointProcessor, recoveryPointProcessor, kinesisStreamName, kinesisApplicationName, kinesisConfig,
-                checkPointMgr, noOfRestarts, kinesisConsumerRegion, null);
+                checkPointMgr, noOfRestarts, kinesisConsumerRegion, kinesisConsumerTracebackSeconds, null);
     }
 
     public KinesisConsumer(MetricRegistry metricRegistry, PointProcessor pointProcessor, PointProcessor recoveryPointProcessor,
                            String kinesisStreamName, String kinesisApplicationName,
                            KinesisConfig kinesisConfig, CheckPointMgr<Date> checkPointMgr,
                            Counter noOfRestarts, String kinesisConsumerRegion,
+                           int kinesisConsumerTracebackSeconds,
                            String overrideKinesisEndpoint) {
         this.metricRegistry = metricRegistry;
         this.pointProcessor = Preconditions.checkNotNull(pointProcessor);
@@ -92,6 +95,7 @@ public KinesisConsumer(MetricRegistry metricRegistry, PointProcessor pointProces
         this.checkPointMgr = checkPointMgr;
         this.noOfRestarts = noOfRestarts;
         this.kinesisConsumerRegion = kinesisConsumerRegion;
+        this.kinesisConsumerTracebackSeconds = kinesisConsumerTracebackSeconds;
         this.overrideKinesisEndpoint = overrideKinesisEndpoint;
         log.info("Kinesis consumer started");
         this.start();
@@ -155,6 +159,10 @@ public void run () {
                 // Map v1 withFailoverTimeMillis -> v2/3 failoverTimeMillis
                 LeaseManagementConfig leaseManagementConfig = configsBuilder.leaseManagementConfig()
                         .failoverTimeMillis(kinesisConfig.getLeaseExpirationTimeInSecs() * 1000L);
+                // Since v2 will create a new DynamoDB table, we need to trace back the initial position
+                InitialPositionInStreamExtended initialPositionInStreamExtended =
+                        InitialPositionInStreamExtended.newInitialPositionAtTimestamp(
+                                new Date(Clock.systemUTC().millis() - kinesisConsumerTracebackSeconds * 1000L));
                 worker = new Scheduler(
                         configsBuilder.checkpointConfig(),
                         configsBuilder.coordinatorConfig(),
@@ -163,9 +171,8 @@ public void run () {
                         configsBuilder.metricsConfig(),
                         configsBuilder.processorConfig(),
                         configsBuilder.retrievalConfig()
-                                .initialPositionInStreamExtended(InitialPositionInStreamExtended.newInitialPosition(InitialPositionInStream.LATEST))
-                                .retrievalSpecificConfig(pollingConfig)
-                );
+                                .initialPositionInStreamExtended(initialPositionInStreamExtended)
+                                .retrievalSpecificConfig(pollingConfig));
                 log.info("KCL v2 Scheduler started with app {}, stream {}, workerId {}", kinesisApplicationName, kinesisStreamName, workerId);
                 worker.run();
             } catch (Throwable t) {
diff --git a/carbonj.service/src/main/java/com/demandware/carbonj/service/engine/cfgCarbonJ.java b/carbonj.service/src/main/java/com/demandware/carbonj/service/engine/cfgCarbonJ.java
@@ -210,6 +210,9 @@ public class cfgCarbonJ
     @Value("${carbonj.env:dev}")
     private String carbonjEnv;
 
+    @Value("${kinesis.consumer.traceback.seconds:3600}")
+    private int kinesisConsumerTracebackSeconds;
+
     @Bean
     public RestTemplate restTemplate() {
         return new RestTemplate();
@@ -427,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);
+                    nsCounter, dataDir == null ? null : new File(dataDir, "index-name-sync"), carbonjEnv, kinesisConsumerTracebackSeconds);
             s.scheduleWithFixedDelay( consumer::reload, 15, 30, TimeUnit.SECONDS );
             if (syncSecondaryDb) {
                 s.scheduleWithFixedDelay( consumer::syncNamespaces, 60, 60, TimeUnit.SECONDS );
diff --git a/carbonj.service/src/test/java/com/demandware/carbonj/service/engine/TestConsumers.java b/carbonj.service/src/test/java/com/demandware/carbonj/service/engine/TestConsumers.java
@@ -38,7 +38,7 @@ public void testConsumers() throws Exception {
                 1, checkPointDir, 60, 60, "recoveryProvider", 1, 1, true);
         FileCheckPointMgr checkPointMgr = new FileCheckPointMgr(checkPointDir, 5);
         Consumers consumers = new Consumers(metricRegistry, new PointProcessorMock(), new PointProcessorMock(),
-                rulesFile, kinesisConfig, checkPointMgr, Region.US_EAST_1.id(), new NamespaceCounter(metricRegistry, 60), new File("/tmp/sync"), "test");
+                rulesFile, kinesisConfig, checkPointMgr, Region.US_EAST_1.id(), new NamespaceCounter(metricRegistry, 60), new File("/tmp/sync"), "test", 60);
         new KinesisRecordProcessorFactory(metricRegistry, new PointProcessorMock(), kinesisConfig, "test-stream", checkPointMgr);
         consumers.dumpStats();
         consumers.syncNamespaces();
@@ -54,7 +54,7 @@ public void testConsumersNoAggregation() throws Exception {
                 1, checkPointDir, 60, 60, "recoveryProvider", 1, 1, false);
         FileCheckPointMgr checkPointMgr = new FileCheckPointMgr(checkPointDir, 5);
         Consumers consumers = new Consumers(metricRegistry, new PointProcessorMock(), new PointProcessorMock(),
-                rulesFile, kinesisConfig, checkPointMgr, Region.US_EAST_1.id(), new NamespaceCounter(metricRegistry, 60), new File("/tmp/sync"), "test");
+                rulesFile, kinesisConfig, checkPointMgr, Region.US_EAST_1.id(), new NamespaceCounter(metricRegistry, 60), new File("/tmp/sync"), "test", 60);
         new KinesisRecordProcessorFactory(metricRegistry, new PointProcessorMock(), kinesisConfig, "test-stream", checkPointMgr);
         consumers.dumpStats();
         consumers.syncNamespaces();
diff --git a/carbonj.service/src/test/java/com/demandware/carbonj/service/engine/TestKinesisConsumer.java b/carbonj.service/src/test/java/com/demandware/carbonj/service/engine/TestKinesisConsumer.java
@@ -93,7 +93,7 @@ public void test() throws Exception {
         PointProcessorMock pointProcessor = new PointProcessorMock();
         KinesisConsumer kinesisConsumer = new KinesisConsumer(metricRegistry, pointProcessor, pointProcessor,
                 STREAM_NAME, STREAM_NAME + "-app", kinesisConfig, checkPointMgr, metricRegistry.counter("kinesis-consumer-counter"),
-                Region.US_EAST_1.id(), localstack.getEndpointOverride(KINESIS).toString());
+                Region.US_EAST_1.id(), 60, localstack.getEndpointOverride(KINESIS).toString());
         Thread.sleep(40000);
         log.info("Start ingesting data points ...");
         int current = (int) (System.currentTimeMillis() / 1000);
