KAFKA-18826: Add global thread metrics #18953

bbejeck · 2025-02-18T18:35:08Z

When implementing KIP-1091 metrics for the Global Stream thread was overlooked. This ticket adds the Global Thread metrics so they are available via the KIP-1076 process of adding external Kafka metrics.

The existing integration test has been updated to confirm GlobalThread metrics are sent via the broker plugin.

bbejeck · 2025-02-18T19:29:42Z

...src/test/java/org/apache/kafka/streams/integration/KafkaStreamsTelemetryIntegrationTest.java

@@ -144,6 +157,51 @@ public void tearDown() throws Exception {
        if (!streamsSecondApplicationProperties.isEmpty()) {
            IntegrationTestUtils.purgeLocalStreamsState(streamsSecondApplicationProperties);
        }
+        if (globalStoreIterator != null) {
+            globalStoreIterator.close();


Required from test - explaination below adding the global store

bbejeck · 2025-02-18T19:32:05Z

...src/test/java/org/apache/kafka/streams/integration/KafkaStreamsTelemetryIntegrationTest.java

+
+    @ParameterizedTest
+    @ValueSource(strings = {"INFO", "DEBUG", "TRACE"})
+    public void shouldPushGlobalThreadMetricsToBroker(final String recordingLevel) throws Exception {


New test for validating broker plugin emits the global thread metrics. I thought of updating the existing broker plugin test, but it's busy enough as is.

bbejeck · 2025-02-18T19:32:28Z

...src/test/java/org/apache/kafka/streams/integration/KafkaStreamsTelemetryIntegrationTest.java


        try (final KafkaStreams streams = new KafkaStreams(topology, streamsApplicationProperties)) {
            IntegrationTestUtils.startApplicationAndWaitUntilRunning(streams);

            final List<MetricName> streamsThreadMetrics = streams.metrics().values().stream().map(Metric::metricName)
-                    .filter(metricName -> metricName.tags().containsKey("thread-id")).collect(Collectors.toList());
+                    .filter(metricName -> metricName.tags().containsKey("thread-id")).toList();


side cleanup

bbejeck · 2025-02-18T19:32:44Z

...src/test/java/org/apache/kafka/streams/integration/KafkaStreamsTelemetryIntegrationTest.java


            final List<MetricName> streamsClientMetrics = streams.metrics().values().stream().map(Metric::metricName)
-                    .filter(metricName -> metricName.group().equals("stream-metrics")).collect(Collectors.toList());
+                    .filter(metricName -> metricName.group().equals("stream-metrics")).toList();


side cleanup

bbejeck · 2025-02-18T19:33:00Z

...src/test/java/org/apache/kafka/streams/integration/KafkaStreamsTelemetryIntegrationTest.java

-            final List<MetricName> consumerPassedStreamThreadMetricNames = INTERCEPTING_CONSUMERS.get(FIRST_INSTANCE_CLIENT).passedMetrics.stream().map(KafkaMetric::metricName).collect(Collectors.toList());
-            final List<MetricName> adminPassedStreamClientMetricNames = INTERCEPTING_ADMIN_CLIENTS.get(FIRST_INSTANCE_CLIENT).passedMetrics.stream().map(KafkaMetric::metricName).collect(Collectors.toList());
+            final List<MetricName> consumerPassedStreamThreadMetricNames = INTERCEPTING_CONSUMERS.get(FIRST_INSTANCE_CLIENT).passedMetrics.stream().map(KafkaMetric::metricName).toList();
+            final List<MetricName> adminPassedStreamClientMetricNames = INTERCEPTING_ADMIN_CLIENTS.get(FIRST_INSTANCE_CLIENT).passedMetrics.stream().map(KafkaMetric::metricName).toList();


side cleanup

bbejeck · 2025-02-18T19:35:41Z

...src/test/java/org/apache/kafka/streams/integration/KafkaStreamsTelemetryIntegrationTest.java

@@ -350,10 +408,10 @@ public void passedMetricsShouldNotLeakIntoClientMetrics() throws Exception {
            IntegrationTestUtils.startApplicationAndWaitUntilRunning(streams);

            final List<MetricName> streamsThreadMetrics = streams.metrics().values().stream().map(Metric::metricName)
-                    .filter(metricName -> metricName.tags().containsKey("thread-id")).collect(Collectors.toList());
+                    .filter(metricName -> metricName.tags().containsKey("thread-id")).toList();


side cleanup here and the next

bbejeck · 2025-02-18T19:47:22Z

...src/test/java/org/apache/kafka/streams/integration/KafkaStreamsTelemetryIntegrationTest.java

@@ -525,7 +619,7 @@ public void exportMetrics(final AuthorizableRequestContext context, final Client
                        .stream()
                        .flatMap(rm -> rm.getScopeMetricsList().stream())
                        .flatMap(sm -> sm.getMetricsList().stream())
-                        .map(metric -> metric.getGauge())
+                        .map(org.apache.kafka.shaded.io.opentelemetry.proto.metrics.v1.Metric::getGauge)


side cleanup as pointed out by IntelliJ not sure using a method handle here and below is better than using a lambda in this case due to needing the fully qualified name

The change LGTM. Little bit annoying that we have naming "conflict" for Metric but using method reference seems better.

bbejeck · 2025-02-18T19:51:57Z

.../java/org/apache/kafka/streams/internals/metrics/StreamsThreadMetricsDelegatingReporter.java

@@ -37,10 +38,10 @@ public class StreamsThreadMetricsDelegatingReporter implements MetricsReporter {
    private final String stateUpdaterThreadId;


-    public StreamsThreadMetricsDelegatingReporter(final Consumer<byte[], byte[]> consumer, final String threadId, final String stateUpdaterThreadId) {
+    public StreamsThreadMetricsDelegatingReporter(final Consumer<byte[], byte[]> consumer, final String threadId, final Optional<String> stateUpdaterThreadId) {


Since the reporting process is exactly the same for the global thread as a stream thread, IMHO it made sense to re-use the existing reporter class. But it seems better to have the stateUpdaterThreadId represented as an Optional to signal that it may not be present

bbejeck · 2025-02-18T20:07:09Z

...src/test/java/org/apache/kafka/streams/integration/KafkaStreamsTelemetryIntegrationTest.java

+                    @Override
+                    public void process(final Record<String, String> record) {
+                        store.put(record.key(), record.value());
+                        globalStoreIterator = store.all();


The store iterator is intentionally not closed here as it needs to be open during the test so the Streams app will emit the org.apache.kafka.stream.state.oldest.iterator.open.since.ms metric that is expected. So the globalStoreIterator is a global variable (pun not intended) so it can be closed in the tearDown method.

Uff... quick a hack. But I guess difficult (impossible) to do differently?

We should add a comment about it though, ie, just c&p your PR comment directly (it's a good explanation, and avoids that somebody spends time tryin to "fix" it, just breaking the test and getting confused, and wasting time).

We might also want to add a @SuppressWarning("resource") to process(...) method? IntelliJ should complain about not using try-with-resource for the iterator?

Regarding @SuppressWarning I wasn't seeing one ,so I didn't add it. If you'd prefer to have on anyway I'll do so.

Seems my IntelliJ warn setting are different. Should be ok w/o it, too.

bbejeck · 2025-02-18T20:13:15Z

.../java/org/apache/kafka/streams/internals/metrics/StreamsThreadMetricsDelegatingReporter.java

        this.consumer = Objects.requireNonNull(consumer);
        this.threadId = Objects.requireNonNull(threadId);
-        this.stateUpdaterThreadId = Objects.requireNonNull(stateUpdaterThreadId);
+        this.stateUpdaterThreadId = stateUpdaterThreadId.orElse("");


Thinking this should be stateUpdaterThreadId.orElse(null); instead - thoughts?

Not sure. Might actually be cleaner to make this.stateUpdaterThreadId and Optional by itself, and update the code accordingly when it's used?

mjsax

Overall LGTM. A few minor comments/questions.

mjsax · 2025-03-01T23:50:52Z

.../java/org/apache/kafka/streams/internals/metrics/StreamsThreadMetricsDelegatingReporter.java

        this.consumer = Objects.requireNonNull(consumer);
        this.threadId = Objects.requireNonNull(threadId);
-        this.stateUpdaterThreadId = Objects.requireNonNull(stateUpdaterThreadId);
+        this.stateUpdaterThreadId = stateUpdaterThreadId.orElse("");


Not sure. Might actually be cleaner to make this.stateUpdaterThreadId and Optional by itself, and update the code accordingly when it's used?

mjsax · 2025-03-01T23:55:21Z

...src/test/java/org/apache/kafka/streams/integration/KafkaStreamsTelemetryIntegrationTest.java

@@ -525,7 +619,7 @@ public void exportMetrics(final AuthorizableRequestContext context, final Client
                        .stream()
                        .flatMap(rm -> rm.getScopeMetricsList().stream())
                        .flatMap(sm -> sm.getMetricsList().stream())
-                        .map(metric -> metric.getGauge())
+                        .map(org.apache.kafka.shaded.io.opentelemetry.proto.metrics.v1.Metric::getGauge)


The change LGTM. Little bit annoying that we have naming "conflict" for Metric but using method reference seems better.

mjsax · 2025-03-01T23:57:00Z

...src/test/java/org/apache/kafka/streams/integration/KafkaStreamsTelemetryIntegrationTest.java

+                Stores.inMemoryKeyValueStore("iq-test-store"),
+                Serdes.String(),
+                Serdes.String()
+        ),


Formatting seems to be off? I would recommend

builder.addGlobalStore( Stores.keyValueStoreBuilder( Stores.inMemoryKeyValueStore("iq-test-store"), Serdes.String(), Serdes.String() ), globalStoreTopic, [...] ):

mjsax · 2025-03-02T00:02:29Z

...src/test/java/org/apache/kafka/streams/integration/KafkaStreamsTelemetryIntegrationTest.java

+                    @Override
+                    public void process(final Record<String, String> record) {
+                        store.put(record.key(), record.value());
+                        globalStoreIterator = store.all();


Uff... quick a hack. But I guess difficult (impossible) to do differently?

We should add a comment about it though, ie, just c&p your PR comment directly (it's a good explanation, and avoids that somebody spends time tryin to "fix" it, just breaking the test and getting confused, and wasting time).

We might also want to add a @SuppressWarning("resource") to process(...) method? IntelliJ should complain about not using try-with-resource for the iterator?

...src/test/java/org/apache/kafka/streams/integration/KafkaStreamsTelemetryIntegrationTest.java

bbejeck · 2025-03-24T16:54:37Z

@mjsax comments addressed

.../java/org/apache/kafka/streams/internals/metrics/StreamsThreadMetricsDelegatingReporter.java

...src/test/java/org/apache/kafka/streams/integration/KafkaStreamsTelemetryIntegrationTest.java

Integrates global thread-specific metrics reporting via `StreamsThreadMetricsDelegatingReporter`. Includes unit and integration tests to validate telemetry push for global stores. Refactors topology creation to conditionally include global stores for testing purposes.

bbejeck · 2025-03-27T17:59:27Z

@mjsax comments addressed

mjsax

Thanks Bill. LGTM.

bbejeck · 2025-04-02T13:18:16Z

Merged #18953 into trunk

When implementing [KIP-1091](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1091%3A+Improved+Kafka+Streams+operator+metrics) metrics for the Global Stream thread was overlooked. This ticket adds the Global Thread metrics so they are available via the KIP-1076 process of adding external Kafka metrics. The existing integration test has been updated to confirm GlobalThread metrics are sent via the broker plugin. Reviewers: Matthias Sax <[email protected]>

bbejeck requested a review from mjsax February 18, 2025 18:35

github-actions bot added the streams label Feb 18, 2025

bbejeck commented Feb 18, 2025

View reviewed changes

mjsax reviewed Mar 2, 2025

View reviewed changes

bbejeck force-pushed the Add_global_thread_metrics branch from 441faac to d77afd8 Compare March 24, 2025 16:47

mjsax reviewed Mar 27, 2025

View reviewed changes

.../java/org/apache/kafka/streams/internals/metrics/StreamsThreadMetricsDelegatingReporter.java Outdated Show resolved Hide resolved

mjsax reviewed Mar 27, 2025

View reviewed changes

...src/test/java/org/apache/kafka/streams/integration/KafkaStreamsTelemetryIntegrationTest.java Outdated Show resolved Hide resolved

mjsax reviewed Mar 27, 2025

View reviewed changes

...src/test/java/org/apache/kafka/streams/integration/KafkaStreamsTelemetryIntegrationTest.java Outdated Show resolved Hide resolved

mjsax reviewed Mar 27, 2025

View reviewed changes

...src/test/java/org/apache/kafka/streams/integration/KafkaStreamsTelemetryIntegrationTest.java Outdated Show resolved Hide resolved

bbejeck added 5 commits March 27, 2025 13:03

Side cleanup, update to keep iterator open for last open metric

bcb31fd

Checkstyle fixes

4660b02

Address review comments

bf68f33

Updates per comments

a9ea85e

bbejeck force-pushed the Add_global_thread_metrics branch from d77afd8 to a9ea85e Compare March 27, 2025 17:38

checkstyle

ec79e97

mjsax approved these changes Apr 2, 2025

View reviewed changes

bbejeck merged commit 87a299a into apache:trunk Apr 2, 2025
21 of 22 checks passed

bbejeck deleted the Add_global_thread_metrics branch April 2, 2025 13:18

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-18826: Add global thread metrics #18953

KAFKA-18826: Add global thread metrics #18953

bbejeck commented Feb 18, 2025 •

edited

Loading

bbejeck Feb 18, 2025

bbejeck Feb 18, 2025

bbejeck Feb 18, 2025

bbejeck Feb 18, 2025

bbejeck Feb 18, 2025

bbejeck Feb 18, 2025

bbejeck Feb 18, 2025

mjsax Mar 1, 2025

bbejeck Feb 18, 2025

bbejeck Feb 18, 2025

mjsax Mar 2, 2025

bbejeck Mar 24, 2025

mjsax Mar 27, 2025

bbejeck Feb 18, 2025

mjsax Mar 1, 2025

mjsax left a comment

mjsax Mar 1, 2025

mjsax Mar 1, 2025

mjsax Mar 1, 2025

mjsax Mar 2, 2025

bbejeck commented Mar 24, 2025

bbejeck commented Mar 27, 2025

mjsax left a comment

bbejeck commented Apr 2, 2025

KAFKA-18826: Add global thread metrics #18953

KAFKA-18826: Add global thread metrics #18953

Conversation

bbejeck commented Feb 18, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mjsax left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bbejeck commented Mar 24, 2025

bbejeck commented Mar 27, 2025

mjsax left a comment

Choose a reason for hiding this comment

bbejeck commented Apr 2, 2025

bbejeck commented Feb 18, 2025 •

edited

Loading