From fcdf57a7f9d01f7b2bb2aad6db42808072ef8274 Mon Sep 17 00:00:00 2001 From: Heng Cui Date: Mon, 9 Jun 2025 12:42:11 +0200 Subject: [PATCH 1/2] update state and restore related loggers --- .../CandyflossKStreamsApplication.java | 6 ++ .../config/JsonKStreamApplicationConfig.java | 3 +- .../MetricLogger.java} | 4 +- .../candyfloss/monitors/RestoreLogger.java | 50 ++++++++++++++++ .../candyfloss/monitors/StateLogger.java | 57 +++++++++++++++++++ 5 files changed, 117 insertions(+), 3 deletions(-) rename candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/{config/CFMetricReporter.java => monitors/MetricLogger.java} (94%) create mode 100644 candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/monitors/RestoreLogger.java create mode 100644 candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/monitors/StateLogger.java diff --git a/candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/CandyflossKStreamsApplication.java b/candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/CandyflossKStreamsApplication.java index ea7eff2..f8857fd 100644 --- a/candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/CandyflossKStreamsApplication.java +++ b/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"); diff --git a/candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/config/JsonKStreamApplicationConfig.java b/candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/config/JsonKStreamApplicationConfig.java index 2f16f5d..9b367e4 100644 --- a/candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/config/JsonKStreamApplicationConfig.java +++ b/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; diff --git a/candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/config/CFMetricReporter.java b/candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/monitors/MetricLogger.java similarity index 94% rename from candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/config/CFMetricReporter.java rename to candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/monitors/MetricLogger.java index a14a7d9..6f758e7 100644 --- a/candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/config/CFMetricReporter.java +++ b/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 metricList) { metricList.forEach(this::metricChange); diff --git a/candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/monitors/RestoreLogger.java b/candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/monitors/RestoreLogger.java new file mode 100644 index 0000000..f6b4743 --- /dev/null +++ b/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); + } +} diff --git a/candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/monitors/StateLogger.java b/candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/monitors/StateLogger.java new file mode 100644 index 0000000..756ea1a --- /dev/null +++ b/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); + } +} From 122c580a30627f93e4ec4ca4abe5b7eb8289b44e Mon Sep 17 00:00:00 2001 From: Heng Cui Date: Mon, 9 Jun 2025 12:42:46 +0200 Subject: [PATCH 2/2] update kafka streams probing rebalance interval --- candyfloss/src/main/resources/application.dev.conf | 1 + candyfloss/src/main/resources/application.prod.conf | 1 + candyfloss/src/main/resources/application.test.conf | 1 + 3 files changed, 3 insertions(+) diff --git a/candyfloss/src/main/resources/application.dev.conf b/candyfloss/src/main/resources/application.dev.conf index c726fc6..883165f 100644 --- a/candyfloss/src/main/resources/application.dev.conf +++ b/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 diff --git a/candyfloss/src/main/resources/application.prod.conf b/candyfloss/src/main/resources/application.prod.conf index 8bbf288..4960756 100644 --- a/candyfloss/src/main/resources/application.prod.conf +++ b/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 diff --git a/candyfloss/src/main/resources/application.test.conf b/candyfloss/src/main/resources/application.test.conf index 9a96ed8..cf2ef28 100644 --- a/candyfloss/src/main/resources/application.test.conf +++ b/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