KAFKA-17561: KIP-1091 add operator metrics #17820

bbejeck · 2024-11-14T20:53:34Z

Implementation of KIP-1091 adding operator metrics to Kafka Streams

Updated existing tests to validate added metrics

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

bbejeck · 2024-11-14T20:55:39Z

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

@@ -180,15 +180,16 @@ public void shouldPushMetricsToBroker(final String recordingLevel) throws Except
                        final String name = mn.name().replace('-', '.');
                        final String group = mn.group().replace("-metrics", "").replace('-', '.');
                        return "org.apache.kafka." + group + "." + name;
-                    }).sorted().collect(Collectors.toList());
+                    }).filter(name -> !name.equals("org.apache.kafka.stream.thread.state"))// telemetry reporter filters out string metrics
+                    .sorted().collect(Collectors.toList());


Telemetry filters out string metrics, so I remove it here from the expected metrics list

I do not understand this. According to the KIP and the code in ThreadMetrics the type of the metric is numeric.

We also added a state metric for stream threads that reports the current state of the thread as a string.

mjsax

Made a pass. Might be good if @cadonna could take a look too.

mjsax · 2024-11-15T01:16:36Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

+        } else if (recordingLevelString.equals("DEBUG")) {
+            recordingLevel = 1;
+        } else {
+            // Must be TRACE level


Might be better to use another if and final else throw an exception as safe guard? Would also highlight that we need to update this code, in case we add a new level?

mjsax · 2024-11-15T01:17:14Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

+    private int calculateMetricsRecordingLevel() {
+        final int recordingLevel;
+        final String recordingLevelString = applicationConfigs.getString(METRICS_RECORDING_LEVEL_CONFIG);
+        if (recordingLevelString.equals("INFO")) {


Why not use a switch statement?

I thought Java 11 had switch expressions but it's Java 13+ so I was a little disappointed, but now looking at again the code will still benefit from using switch

mjsax · 2024-11-15T01:18:56Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

@@ -614,6 +615,12 @@ public StreamThread(final Time time,
            streamsMetrics,
            time.milliseconds()
        );
+        ThreadMetrics.addThreadStateTelemetryMetric(threadId,


nit: threadId should be in the next line by itself

nit: we usually use 4 spaces indentation.

mjsax · 2024-11-15T01:19:03Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

+        ThreadMetrics.addThreadStateTelemetryMetric(threadId,
+                streamsMetrics,
+                (metricConfig, now) -> this.state().ordinal());
+        ThreadMetrics.addThreadStateMetric(threadId,


mjsax · 2024-11-15T01:22:43Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

+                (metricConfig, now) -> this.state().ordinal());
+        ThreadMetrics.addThreadStateMetric(threadId,
+                streamsMetrics,
+                (metricConfig, now) -> this.state().name().toLowerCase(Locale.getDefault()));


For the corresponding client metric we just use

ClientMetrics.addStateMetric(streamsMetrics, (metricsConfig, now) -> state);

Why so "complicated" -- I am also ok to update the code for the client metric. But both should be the same?

(Or maybe keep ... -> state and add a "fancy" toString() overload to both enum (for client and thread) which model the state?

The KIP specifies lower-case string but I like the other approach so I'll update it.

mjsax · 2024-11-15T01:24:00Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/ThreadMetrics.java

+                                                     final StreamsMetricsImpl streamsMetrics,
+                                                     final Gauge<Integer> threadStateProvider) {
+        streamsMetrics.addThreadLevelMutableMetric(
+                THREAD_STATE,


nit: indention too deep (should only be 4 whitespace, not 8 -- should be an IDE setting)

Ack, I think I need to copy settings from someone this is a common occurance with my PRs

mjsax · 2024-11-15T01:24:21Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/ThreadMetrics.java

+                                            final StreamsMetricsImpl streamsMetrics,
+                                            final Gauge<String> threadStateProvider) {
+        streamsMetrics.addThreadLevelMutableMetric(
+                STATE,


mjsax · 2024-11-15T01:31:26Z

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

            final List<String> actualMetrics = new ArrayList<>(TelemetryPlugin.SUBSCRIBED_METRICS.get(mainConsumerInstanceId));
            assertEquals(expectedMetrics, actualMetrics);

            TestUtils.waitForCondition(() -> !TelemetryPlugin.SUBSCRIBED_METRICS.get(adminInstanceId).isEmpty(),
                    30_000,
                    "Never received subscribed metrics");
            final List<String> actualInstanceMetrics = TelemetryPlugin.SUBSCRIBED_METRICS.get(adminInstanceId);
-            final List<String> expectedInstanceMetrics = Arrays.asList("org.apache.kafka.stream.alive.stream.threads", "org.apache.kafka.stream.failed.stream.threads");
+            final List<String> expectedInstanceMetrics = Arrays.asList("org.apache.kafka.stream.alive.stream.threads", "org.apache.kafka.stream.client.state", "org.apache.kafka.stream.failed.stream.threads", "org.apache.kafka.stream.recording.level");


Nit: line too long. How about

final List<String> expectedInstanceMetrics = Arrays.asList( "org.apache.kafka.stream.client.state", "org.apache.kafka.stream.alive.stream.threads", "org.apache.kafka.stream.failed.stream.threads", "org.apache.kafka.stream.recording.level" );

cadonna

Thanks for the PR, @bbejeck !

Could you please also update MetricsIntegrationTest with the new metrics?

cadonna · 2024-11-15T09:51:25Z

...ms/src/test/java/org/apache/kafka/streams/processor/internals/metrics/ThreadMetricsTest.java

+    public void shouldAddThreadStateTelemetryMetric() {
+        final Gauge<Integer> threadStateProvider = (streamsMetrics, startTime) -> StreamThread.State.RUNNING.ordinal();
+        ThreadMetrics.addThreadStateTelemetryMetric(
+                "threadId",


There is a constant for a test thread ID:

Suggested change

"threadId",

THREAD_ID,

cadonna · 2024-11-15T09:53:48Z

...ms/src/test/java/org/apache/kafka/streams/processor/internals/metrics/ThreadMetricsTest.java

+        verify(streamsMetrics).addThreadLevelMutableMetric(
+                "thread-state",
+                "The current state of the thread",
+                "threadId",


Suggested change

"threadId",

THREAD_ID,

cadonna · 2024-11-15T09:54:00Z

...ms/src/test/java/org/apache/kafka/streams/processor/internals/metrics/ThreadMetricsTest.java

+    public void shouldAddThreadStateJMXMetric() {
+        final Gauge<String> threadStateProvider = (streamsMetrics, startTime) -> StreamThread.State.RUNNING.name().toLowerCase(Locale.getDefault());
+        ThreadMetrics.addThreadStateMetric(
+                "threadId",


Suggested change

"threadId",

THREAD_ID,

cadonna · 2024-11-15T09:54:10Z

...ms/src/test/java/org/apache/kafka/streams/processor/internals/metrics/ThreadMetricsTest.java

+        verify(streamsMetrics).addThreadLevelMutableMetric(
+                "state",
+                "The current state of the thread",
+                "threadId",


Suggested change

"threadId",

THREAD_ID,

cadonna · 2024-11-15T09:55:53Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

@@ -614,6 +615,12 @@ public StreamThread(final Time time,
            streamsMetrics,
            time.milliseconds()
        );
+        ThreadMetrics.addThreadStateTelemetryMetric(threadId,


nit: we usually use 4 spaces indentation.

cadonna · 2024-11-15T09:59:42Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

-            .recordLevel(Sensor.RecordingLevel.forName(config.getString(StreamsConfig.METRICS_RECORDING_LEVEL_CONFIG)))
+            .recordLevel(Sensor.RecordingLevel.forName(config.getString(METRICS_RECORDING_LEVEL_CONFIG)))


Why did you remove the StreamsConfig. prefix? For all other config names, we use the prefix.

Unintended, I let IntelliJ do the driving on that one, I'll revert

cadonna · 2024-11-15T10:13:20Z

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

@@ -180,15 +180,16 @@ public void shouldPushMetricsToBroker(final String recordingLevel) throws Except
                        final String name = mn.name().replace('-', '.');
                        final String group = mn.group().replace("-metrics", "").replace('-', '.');
                        return "org.apache.kafka." + group + "." + name;
-                    }).sorted().collect(Collectors.toList());
+                    }).filter(name -> !name.equals("org.apache.kafka.stream.thread.state"))// telemetry reporter filters out string metrics
+                    .sorted().collect(Collectors.toList());


I do not understand this. According to the KIP and the code in ThreadMetrics the type of the metric is numeric.

cadonna · 2024-11-15T10:16:49Z

...ms/src/test/java/org/apache/kafka/streams/processor/internals/metrics/ThreadMetricsTest.java

+    }
+
+    @Test
+    public void shouldAddThreadStateJMXMetric() {


nit:

Suggested change

public void shouldAddThreadStateJMXMetric() {

public void shouldAddThreadStateJmxMetric() {

bbejeck · 2024-11-15T22:45:54Z

@mjsax and @cadonna - thanks for the reviews! All comments addressed.

mjsax

LGTM. Assuming build passes.

bbejeck added 3 commits November 13, 2024 18:47

Initial commit for adding operator metrics

d757da0

Added all operator metrics, updated integration test

4ace02e

Added tests

d834ea2

github-actions bot added the streams label Nov 14, 2024

bbejeck commented Nov 14, 2024

View reviewed changes

bbejeck requested review from cadonna and mjsax November 14, 2024 20:58

mjsax added the kip Requires or implements a KIP label Nov 15, 2024

mjsax reviewed Nov 15, 2024

View reviewed changes

cadonna reviewed Nov 15, 2024

View reviewed changes

Address review comments

06884bc

mjsax approved these changes Nov 16, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-17561: KIP-1091 add operator metrics #17820

KAFKA-17561: KIP-1091 add operator metrics #17820

bbejeck commented Nov 14, 2024

bbejeck Nov 14, 2024

cadonna Nov 15, 2024

bbejeck Nov 15, 2024

mjsax left a comment

mjsax Nov 15, 2024

mjsax Nov 15, 2024

bbejeck Nov 15, 2024

mjsax Nov 15, 2024

cadonna Nov 15, 2024

mjsax Nov 15, 2024

mjsax Nov 15, 2024

bbejeck Nov 15, 2024

mjsax Nov 15, 2024

bbejeck Nov 15, 2024

mjsax Nov 15, 2024

mjsax Nov 15, 2024

cadonna left a comment

cadonna Nov 15, 2024

cadonna Nov 15, 2024

cadonna Nov 15, 2024

cadonna Nov 15, 2024

cadonna Nov 15, 2024

cadonna Nov 15, 2024

bbejeck Nov 15, 2024

cadonna Nov 15, 2024

cadonna Nov 15, 2024

bbejeck commented Nov 15, 2024

mjsax left a comment

		.recordLevel(Sensor.RecordingLevel.forName(config.getString(StreamsConfig.METRICS_RECORDING_LEVEL_CONFIG)))
		.recordLevel(Sensor.RecordingLevel.forName(config.getString(METRICS_RECORDING_LEVEL_CONFIG)))

	public void shouldAddThreadStateJMXMetric() {
	public void shouldAddThreadStateJmxMetric() {

KAFKA-17561: KIP-1091 add operator metrics #17820

Are you sure you want to change the base?

KAFKA-17561: KIP-1091 add operator metrics #17820

Conversation

bbejeck commented Nov 14, 2024

Committer Checklist (excluded from commit message)

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

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

cadonna 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

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 Nov 15, 2024

mjsax left a comment

Choose a reason for hiding this comment