additional kafka streams state and store loggers

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

@@ -4,6 +4,8 @@
 import com.swisscom.daisy.cosmos.candyfloss.config.MetricRegistryConfig;
 import com.swisscom.daisy.cosmos.candyfloss.config.exceptions.InvalidConfigurations;
 import com.swisscom.daisy.cosmos.candyfloss.messages.ValueErrorMessage;
+import com.swisscom.daisy.cosmos.candyfloss.monitors.RestoreLogger;
+import com.swisscom.daisy.cosmos.candyfloss.monitors.StateLogger;
 import com.swisscom.daisy.cosmos.candyfloss.processors.*;
 import com.swisscom.daisy.cosmos.candyfloss.transformations.Transformer;
 import com.swisscom.daisy.cosmos.candyfloss.transformations.match.exceptions.InvalidMatchConfiguration;
@@ -70,6 +72,10 @@ public void run() {
 
     try {
       prometheusRegistry.start();
+
+      StateLogger.setListener(kafkaStreams, new StateLogger(kafkaStreams));
+      RestoreLogger.setListener(kafkaStreams, new RestoreLogger());
+
       kafkaStreams.start();
       latch.await();
       logger.info("Cosmos Candyfloss gracefully shutdown");

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

@@ -1,6 +1,7 @@
 package com.swisscom.daisy.cosmos.candyfloss.config;
 
 import com.swisscom.daisy.cosmos.candyfloss.config.exceptions.InvalidConfigurations;
+import com.swisscom.daisy.cosmos.candyfloss.monitors.MetricLogger;
 import com.swisscom.daisy.cosmos.candyfloss.testutils.JsonUtil;
 import com.swisscom.daisy.cosmos.candyfloss.transformations.match.exceptions.InvalidMatchConfiguration;
 import com.typesafe.config.Config;
@@ -46,7 +47,7 @@ public static JsonKStreamApplicationConfig fromConfig(Config config)
     for (var entry : kafkaConfig.entrySet()) {
       kafka.put(entry.getKey(), kafkaConfig.getString(entry.getKey()));
     }
-    kafka.put(StreamsConfig.METRIC_REPORTER_CLASSES_CONFIG, CFMetricReporter.class.getName());
+    kafka.put(StreamsConfig.METRIC_REPORTER_CLASSES_CONFIG, MetricLogger.class.getName());
 
     final String inputTopicName = config.getConfig("kstream").getString("input.topic.name");
     final InputType inputType;

candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/config/CFMetricReporter.java → candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/monitors/MetricLogger.java

@@ -1,4 +1,4 @@
-package com.swisscom.daisy.cosmos.candyfloss.config;
+package com.swisscom.daisy.cosmos.candyfloss.monitors;
 
 import io.micrometer.core.instrument.*;
 import java.util.List;
@@ -7,7 +7,7 @@
 import org.apache.kafka.common.metrics.KafkaMetric;
 import org.apache.kafka.common.metrics.MetricsReporter;
 
-public class CFMetricReporter implements MetricsReporter {
+public class MetricLogger implements MetricsReporter {
   @Override
   public void init(List<KafkaMetric> metricList) {
     metricList.forEach(this::metricChange);

candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/monitors/RestoreLogger.java

@@ -0,0 +1,50 @@
+package com.swisscom.daisy.cosmos.candyfloss.monitors;
+
+import io.micrometer.core.instrument.Metrics;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.processor.StateRestoreListener;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RestoreLogger implements StateRestoreListener {
+  private static final Logger logger = LoggerFactory.getLogger(RestoreLogger.class);
+  private static final String cfRestoreCounter = "cf_restore_records";
+  private static final String metricTagThread = "thread";
+  private static final String metricTagPartition = "partition";
+
+  @Override
+  public void onRestoreStart(
+      TopicPartition changelogPartition, String storeName, long beginOffset, long endOffset) {
+    logger.info(
+        "START - State restore from changelog partition: {}, from offset: {} to offset: {}",
+        changelogPartition.partition(),
+        beginOffset,
+        endOffset);
+  }
+
+  @Override
+  public void onBatchRestored(
+      TopicPartition changelogPartition, String storeName, long offset, long numRestored) {
+    String partition = String.valueOf(changelogPartition.partition());
+    String threadName = Thread.currentThread().getName();
+
+    Metrics.globalRegistry
+        .counter(cfRestoreCounter, metricTagThread, threadName, metricTagPartition, partition)
+        .increment(numRestored);
+  }
+
+  @Override
+  public void onRestoreEnd(
+      TopicPartition changelogPartition, String storeName, long totalRestored) {
+    logger.info(
+        "END - State restore from changelog partition: {}, total restored: {}",
+        changelogPartition.partition(),
+        totalRestored);
+  }
+
+  public static void setListener(KafkaStreams kafkaStreams, StateRestoreListener listener) {
+    logger.info("Enable Candyfloss restore monitoring (logging and metrics)");
+    kafkaStreams.setGlobalStateRestoreListener(listener);
+  }
+}

candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/monitors/StateLogger.java

@@ -0,0 +1,57 @@
+package com.swisscom.daisy.cosmos.candyfloss.monitors;
+
+import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.KafkaStreams.StateListener;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class StateLogger implements StateListener {
+  private final KafkaStreams streams;
+
+  public StateLogger(KafkaStreams ks) {
+    streams = ks;
+  }
+
+  private static final Logger logger = LoggerFactory.getLogger(StateLogger.class);
+
+  private void displayCurrentPartitionInfo() {
+    streams
+        .metadataForLocalThreads()
+        .forEach(
+            metadata -> {
+              if (!metadata.activeTasks().isEmpty()) {
+                metadata
+                    .activeTasks()
+                    .forEach(
+                        partition ->
+                            logger.info(
+                                "[{}] assigned partition: {}",
+                                Thread.currentThread().getName(),
+                                partition.topicPartitions()));
+              }
+            });
+  }
+
+  @Override
+  public void onChange(KafkaStreams.State newState, KafkaStreams.State oldState) {
+    logger.info(
+        "[Candyfloss] State transition from oldState: {} to newState: {}", oldState, newState);
+    switch (newState) {
+      case RUNNING:
+        displayCurrentPartitionInfo();
+        break;
+      case CREATED:
+      case REBALANCING:
+      case PENDING_SHUTDOWN:
+      case NOT_RUNNING:
+      case ERROR:
+      default:
+        break;
+    }
+  }
+
+  public static void setListener(KafkaStreams kafkaStreams, StateListener listener) {
+    logger.info("Enable Candyfloss state listener");
+    kafkaStreams.setStateListener(listener);
+  }
+}

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

@@ -12,6 +12,7 @@ kafka {
   compression.type = gzip
   replication.factor = 2
   num.stream.threads = 2
+  probing.rebalance.interval.ms = 120000
 }
 kstream {
   input.topic.name = dev.device-json-raw-input

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

@@ -12,6 +12,7 @@ kafka {
   compression.type = gzip
   replication.factor = 2
   num.stream.threads = 2
+  probing.rebalance.interval.ms = 120000
 }
 kstream {
   input.topic.name = prod.device-json-raw-input

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

@@ -12,6 +12,7 @@ kafka {
   compression.type = gzip
   replication.factor = 2
   num.stream.threads = 2
+  probing.rebalance.interval.ms = 120000
 }
 kstream {
   input.topic.name = test.device-json-raw-input