Skip to content

Commit fcdf57a

Browse files
Heng CuiHeng Cui
Heng Cui
authored and
Heng Cui
committed
update state and restore related loggers
1 parent dd3e387 commit fcdf57a

5 files changed

Lines changed: 117 additions & 3 deletions

File tree

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

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,8 @@
44
import com.swisscom.daisy.cosmos.candyfloss.config.MetricRegistryConfig;
55
import com.swisscom.daisy.cosmos.candyfloss.config.exceptions.InvalidConfigurations;
66
import com.swisscom.daisy.cosmos.candyfloss.messages.ValueErrorMessage;
7+
import com.swisscom.daisy.cosmos.candyfloss.monitors.RestoreLogger;
8+
import com.swisscom.daisy.cosmos.candyfloss.monitors.StateLogger;
79
import com.swisscom.daisy.cosmos.candyfloss.processors.*;
810
import com.swisscom.daisy.cosmos.candyfloss.transformations.Transformer;
911
import com.swisscom.daisy.cosmos.candyfloss.transformations.match.exceptions.InvalidMatchConfiguration;
@@ -70,6 +72,10 @@ public void run() {
7072

7173
try {
7274
prometheusRegistry.start();
75+
76+
StateLogger.setListener(kafkaStreams, new StateLogger(kafkaStreams));
77+
RestoreLogger.setListener(kafkaStreams, new RestoreLogger());
78+
7379
kafkaStreams.start();
7480
latch.await();
7581
logger.info("Cosmos Candyfloss gracefully shutdown");

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

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
package com.swisscom.daisy.cosmos.candyfloss.config;
22

33
import com.swisscom.daisy.cosmos.candyfloss.config.exceptions.InvalidConfigurations;
4+
import com.swisscom.daisy.cosmos.candyfloss.monitors.MetricLogger;
45
import com.swisscom.daisy.cosmos.candyfloss.testutils.JsonUtil;
56
import com.swisscom.daisy.cosmos.candyfloss.transformations.match.exceptions.InvalidMatchConfiguration;
67
import com.typesafe.config.Config;
@@ -46,7 +47,7 @@ public static JsonKStreamApplicationConfig fromConfig(Config config)
4647
for (var entry : kafkaConfig.entrySet()) {
4748
kafka.put(entry.getKey(), kafkaConfig.getString(entry.getKey()));
4849
}
49-
kafka.put(StreamsConfig.METRIC_REPORTER_CLASSES_CONFIG, CFMetricReporter.class.getName());
50+
kafka.put(StreamsConfig.METRIC_REPORTER_CLASSES_CONFIG, MetricLogger.class.getName());
5051

5152
final String inputTopicName = config.getConfig("kstream").getString("input.topic.name");
5253
final InputType inputType;

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
package com.swisscom.daisy.cosmos.candyfloss.config;
1+
package com.swisscom.daisy.cosmos.candyfloss.monitors;
22

33
import io.micrometer.core.instrument.*;
44
import java.util.List;
@@ -7,7 +7,7 @@
77
import org.apache.kafka.common.metrics.KafkaMetric;
88
import org.apache.kafka.common.metrics.MetricsReporter;
99

10-
public class CFMetricReporter implements MetricsReporter {
10+
public class MetricLogger implements MetricsReporter {
1111
@Override
1212
public void init(List<KafkaMetric> metricList) {
1313
metricList.forEach(this::metricChange);

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

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,50 @@
1+
package com.swisscom.daisy.cosmos.candyfloss.monitors;
2+
3+
import io.micrometer.core.instrument.Metrics;
4+
import org.apache.kafka.common.TopicPartition;
5+
import org.apache.kafka.streams.KafkaStreams;
6+
import org.apache.kafka.streams.processor.StateRestoreListener;
7+
import org.slf4j.Logger;
8+
import org.slf4j.LoggerFactory;
9+
10+
public class RestoreLogger implements StateRestoreListener {
11+
private static final Logger logger = LoggerFactory.getLogger(RestoreLogger.class);
12+
private static final String cfRestoreCounter = "cf_restore_records";
13+
private static final String metricTagThread = "thread";
14+
private static final String metricTagPartition = "partition";
15+
16+
@Override
17+
public void onRestoreStart(
18+
TopicPartition changelogPartition, String storeName, long beginOffset, long endOffset) {
19+
logger.info(
20+
"START - State restore from changelog partition: {}, from offset: {} to offset: {}",
21+
changelogPartition.partition(),
22+
beginOffset,
23+
endOffset);
24+
}
25+
26+
@Override
27+
public void onBatchRestored(
28+
TopicPartition changelogPartition, String storeName, long offset, long numRestored) {
29+
String partition = String.valueOf(changelogPartition.partition());
30+
String threadName = Thread.currentThread().getName();
31+
32+
Metrics.globalRegistry
33+
.counter(cfRestoreCounter, metricTagThread, threadName, metricTagPartition, partition)
34+
.increment(numRestored);
35+
}
36+
37+
@Override
38+
public void onRestoreEnd(
39+
TopicPartition changelogPartition, String storeName, long totalRestored) {
40+
logger.info(
41+
"END - State restore from changelog partition: {}, total restored: {}",
42+
changelogPartition.partition(),
43+
totalRestored);
44+
}
45+
46+
public static void setListener(KafkaStreams kafkaStreams, StateRestoreListener listener) {
47+
logger.info("Enable Candyfloss restore monitoring (logging and metrics)");
48+
kafkaStreams.setGlobalStateRestoreListener(listener);
49+
}
50+
}

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

Lines changed: 57 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,57 @@
1+
package com.swisscom.daisy.cosmos.candyfloss.monitors;
2+
3+
import org.apache.kafka.streams.KafkaStreams;
4+
import org.apache.kafka.streams.KafkaStreams.StateListener;
5+
import org.slf4j.Logger;
6+
import org.slf4j.LoggerFactory;
7+
8+
public class StateLogger implements StateListener {
9+
private final KafkaStreams streams;
10+
11+
public StateLogger(KafkaStreams ks) {
12+
streams = ks;
13+
}
14+
15+
private static final Logger logger = LoggerFactory.getLogger(StateLogger.class);
16+
17+
private void displayCurrentPartitionInfo() {
18+
streams
19+
.metadataForLocalThreads()
20+
.forEach(
21+
metadata -> {
22+
if (!metadata.activeTasks().isEmpty()) {
23+
metadata
24+
.activeTasks()
25+
.forEach(
26+
partition ->
27+
logger.info(
28+
"[{}] assigned partition: {}",
29+
Thread.currentThread().getName(),
30+
partition.topicPartitions()));
31+
}
32+
});
33+
}
34+
35+
@Override
36+
public void onChange(KafkaStreams.State newState, KafkaStreams.State oldState) {
37+
logger.info(
38+
"[Candyfloss] State transition from oldState: {} to newState: {}", oldState, newState);
39+
switch (newState) {
40+
case RUNNING:
41+
displayCurrentPartitionInfo();
42+
break;
43+
case CREATED:
44+
case REBALANCING:
45+
case PENDING_SHUTDOWN:
46+
case NOT_RUNNING:
47+
case ERROR:
48+
default:
49+
break;
50+
}
51+
}
52+
53+
public static void setListener(KafkaStreams kafkaStreams, StateListener listener) {
54+
logger.info("Enable Candyfloss state listener");
55+
kafkaStreams.setStateListener(listener);
56+
}
57+
}

0 commit comments

Comments
 (0)