delete statestore if ts far-away reordered

Author: Heng Cui

candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/CandyflossKStreamsApplication.java

@@ -221,6 +221,7 @@ public Topology buildTopology() {
                     new CounterNormalizationProcessor(
                         config.getPipeline(),
                         config.getStateStoreName(),
+                        config.getStateStoreCutoffTime(),
                         config.getMaxCounterCacheAge(),
                         config.getIntCounterWrapAroundLimit(),
                         config.getLongCounterWrapAroundLimit(),

candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/config/JsonKStreamApplicationConfig.java

@@ -32,6 +32,7 @@ public enum InputType {
   private final String discardTopicName;
   private final String dlqTopicName;
   private final String stateStoreName;
+  private final long stateStoreCutoffTime;
   private final long maxCounterCacheAge;
   private final int intCounterWrapAroundLimit;
   private final long longCounterWrapAroundLimit;
@@ -58,6 +59,8 @@ public static JsonKStreamApplicationConfig fromConfig(Config config)
     }
     final String discardTopicName = config.getConfig("kstream").getString("discard.topic.name");
     final String dlqTopicName = config.getConfig("kstream").getString("dlq.topic.name");
+    final long stateStoreCutoffTime =
+        config.getConfig("kstream").getLong("normalization.statestore.cutoff.time.ms");
     final String stateStoreName = config.getConfig("kstream").getString("state.store.name");
     final long maxCounterCacheAge =
         config.getConfig("kstream").getLong("state.store.max.counter.cache.age");
@@ -90,6 +93,7 @@ public static JsonKStreamApplicationConfig fromConfig(Config config)
         discardTopicName,
         dlqTopicName,
         stateStoreName,
+        stateStoreCutoffTime,
         maxCounterCacheAge,
         intCounterWrapAroundLimit,
         longCounterWrapAroundLimit,

candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/processors/CounterNormalizationProcessor.java

@@ -41,6 +41,7 @@ public class CounterNormalizationProcessor
           .description("Number of error messages that are discarded to dlq topic")
           .register(Metrics.globalRegistry);
   private final String stateStoreName;
+  private final long stateStoreCutoffTime;
   private final PipelineConfig pipelineConfig;
   private TimestampedKeyValueStore<Bytes, Bytes> stateStore;
   private final long maxCounterCacheAge;
@@ -67,6 +68,7 @@ public class CounterNormalizationProcessor
   public CounterNormalizationProcessor(
       PipelineConfig pipelineConfig,
       String stateStoreName,
+      long stateStoreCutoffTime,
       long maxCounterCacheAge,
       int intCounterWrapAroundLimit,
       long longCounterWrapAroundLimit,
@@ -75,6 +77,7 @@ public CounterNormalizationProcessor(
       Duration scanFrequency) {
     this.pipelineConfig = pipelineConfig;
     this.stateStoreName = stateStoreName;
+    this.stateStoreCutoffTime = stateStoreCutoffTime;
     this.maxCounterCacheAge = maxCounterCacheAge;
     this.intCounterWrapAroundLimit = intCounterWrapAroundLimit;
     this.longCounterWrapAroundLimit = longCounterWrapAroundLimit;
@@ -175,10 +178,17 @@ private void normalizeCounter(
       }
       if (timestamp.toEpochMilli() < savedCounterState.timestamp()) {
         logger.warn(
-            "Received a message with a timestamp {} older than saved in the counter store {}. Message: {}",
+            "Received a message with a timestamp {} older than saved in the counter store {} "
+                + "(timestamp difference: {} ms). Message: {}",
             timestamp.toEpochMilli(),
             savedCounterState.timestamp(),
+            (timestamp.toEpochMilli() - savedCounterState.timestamp()),
             flattenedMessage.getValue().json());
+
+        if (savedCounterState.timestamp() - timestamp.toEpochMilli() > stateStoreCutoffTime) {
+          stateStore.delete(counterKey);
+        }
+
         savedCounterState = null;
       } else {
         // Save the new counterValue to the store

candyfloss/src/main/resources/application.dev.conf

@@ -18,6 +18,7 @@ kstream {
   input.topic.name = dev.device-json-raw-input
   discard.topic.name = dev.candyfloss-processing-discard # All messages that didn't match any step in the pipeline
   dlq.topic.name = dev.candyfloss-processing--dlq # Messages that encountered a Java exception during the processing
+  normalization.statestore.cutoff.time.ms = 1800000 # 30 minutes
   state.store.name = dev.candyfloss-counters-store # Kafka state store name to save the counter values
   state.store.max.counter.cache.age = 900000 // 15 minutes
   state.store.int.counter.wrap.limit = 10000

candyfloss/src/main/resources/application.ietf.dev.conf

@@ -20,6 +20,7 @@ kstream {
   input.topic.name = daisy.dev.device-json-raw
   discard.topic.name = daisy.dev.candyfloss-discard
   dlq.topic.name = daisy.dev.candyfloss-dlq
+  normalization.statestore.cutoff.time.ms = 1800000 # 30 minutes
   state.store.name = candyfloss-counters-store
   state.store.max.counter.cache.age = 900000 // 15 minutes
   state.store.int.counter.wrap.limit = 10000

candyfloss/src/main/resources/application.prod.conf

@@ -18,6 +18,7 @@ kstream {
   input.topic.name = prod.device-json-raw-input
   discard.topic.name = prod.candyfloss-processing-discard # All messages that didn't match any step in the pipeline
   dlq.topic.name = prod.candyfloss-processing-dlq # Messages that encountered a Java exception during the processing
+  normalization.statestore.cutoff.time.ms = 1800000 # 30 minutes
   state.store.name = prod.candyfloss-counters-store # Kafka state store name to save the counter values
   state.store.max.counter.cache.age = 900000 // 15 minutes
   state.store.int.counter.wrap.limit = 10000

candyfloss/src/main/resources/application.test.conf

@@ -19,6 +19,7 @@ kstream {
   discard.topic.name = test.candyfloss-processing-discard # All messages that didn't match any step in the pipeline
   dlq.topic.name = test.candyfloss-processing-dlq # Messages that encountered a Java exception during the processing
   state.store.name = test.candyfloss-counters-store # Kafka state store name to save the counter values
+  normalization.statestore.cutoff.time.ms = 1800000 # 30 minutes
   state.store.max.counter.cache.age = 900000 // 15 minutes
   state.store.int.counter.wrap.limit = 10000
   state.store.long.counter.wrap.limit = 10000000

candyfloss/src/test/java/com/swisscom/daisy/cosmos/candyfloss/processors/CounterNormalizationProcessorTest.java

@@ -96,6 +96,7 @@ private Topology getTestTopology(JsonKStreamApplicationConfig appConf) {
                         appConf.getPipeline(),
                         appConf.getStateStoreName(),
                         appConf.getMaxCounterCacheAge(),
+                        appConf.getStateStoreCutoffTime(),
                         appConf.getIntCounterWrapAroundLimit(),
                         appConf.getLongCounterWrapAroundLimit(),
                         appConf.getCounterWrapAroundTimeMs(),
@@ -429,33 +430,33 @@ void transformCounterNegative()
   void transformCounterOlderMilli()
       throws IOException, JSONException, InvalidConfigurations, InvalidMatchConfiguration {
     setup("counter-normalization/application-milli.conf");
-    var input1 =
-        JsonUtil.readFromInputStream(
-            getClass()
-                .getClassLoader()
-                .getResourceAsStream("counter-normalization/older/input1.json"));
-    var input2 =
-        JsonUtil.readFromInputStream(
-            getClass()
-                .getClassLoader()
-                .getResourceAsStream("counter-normalization/older/input2.json"));
-    var expected1 =
-        JsonUtil.readFromInputStream(
-            getClass()
-                .getClassLoader()
-                .getResourceAsStream("counter-normalization/older/output1.json"));
-    var expected2 =
-        JsonUtil.readFromInputStream(
-            getClass()
-                .getClassLoader()
-                .getResourceAsStream("counter-normalization/older/output2.json"));
+    String resourcePath = "counter-normalization/older/";
+    List<Path> inputFiles = retrieveSortedFilenames(resourcePath, "input");
+    List<Path> outputFiles = retrieveSortedFilenames(resourcePath, "output");
 
-    inputTopic.pipeInput("k1", input1);
-    inputTopic.pipeInput("k1", input2);
+    assertEquals(inputFiles.size(), outputFiles.size());
+    for (var i = 0; i < inputFiles.size(); i++) {
+      var input = inputFiles.get(i).getFileName();
+      var inputString =
+          JsonUtil.readFromInputStream(
+              getClass()
+                  .getClassLoader()
+                  .getResourceAsStream(String.format("%s/%s", resourcePath, input)));
+
+      inputTopic.pipeInput("k1", inputString);
+    }
 
     var processed = outputTopic.readValuesToList();
-    assertEquals(2, processed.size());
-    JSONAssert.assertEquals(expected1, processed.get(0), true);
-    JSONAssert.assertEquals(expected2, processed.get(1), true);
+    assertEquals(inputFiles.size(), processed.size());
+    for (var i = 0; i < outputFiles.size(); i++) {
+      var output = outputFiles.get(i).getFileName();
+      var expectedString =
+          JsonUtil.readFromInputStream(
+              getClass()
+                  .getClassLoader()
+                  .getResourceAsStream(String.format("%s/%s", resourcePath, output)));
+
+      JSONAssert.assertEquals(expectedString, processed.get(i), true);
+    }
   }
 }

candyfloss/src/test/resources/avro-input/application.conf

@@ -16,6 +16,7 @@ kstream {
   discard.topic.name = daisy.dev.avro-input-discard
   dlq.topic.name =daisy.dev.avro-input-dlq
   state.store.name = daisy.dev.avro-input-store
+  normalization.statestore.cutoff.time.ms = 1800000 # 30 minutes
   state.store.max.counter.cache.age = 1000
   state.store.int.counter.wrap.limit = 10000
   state.store.long.counter.wrap.limit = 10000000

candyfloss/src/test/resources/counter-normalization/application-milli.conf

@@ -13,6 +13,7 @@ kstream {
   input.topic.name = daisy.prod.device-json-raw
   discard.topic.name = daisy.dev.device-json-flat
   dlq.topic.name = daisy.dev.device-json-flat-dlq
+  normalization.statestore.cutoff.time.ms = 1800000 # 30 minutes
   state.store.name = daisy.dev.device-counters-store
   state.store.max.counter.cache.age = 10000
   state.store.int.counter.wrap.limit = 10000