KAFKA-17561: KIP-1091 add operator metrics

streams/integration-tests/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());
             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");
             assertEquals(expectedInstanceMetrics, actualInstanceMetrics);
 
             TestUtils.waitForCondition(() -> TelemetryPlugin.processId != null,

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

@@ -110,6 +110,7 @@
 import java.util.function.Consumer;
 import java.util.function.Supplier;
 
+import static org.apache.kafka.streams.StreamsConfig.METRICS_RECORDING_LEVEL_CONFIG;
 import static org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_CLIENT;
 import static org.apache.kafka.streams.internals.ApiUtils.prepareMillisCheckFailMsgPrefix;
 import static org.apache.kafka.streams.internals.ApiUtils.validateMillisecondDuration;
@@ -987,6 +988,8 @@ private KafkaStreams(final TopologyMetadata topologyMetadata,
         ClientMetrics.addApplicationIdMetric(streamsMetrics, applicationConfigs.getString(StreamsConfig.APPLICATION_ID_CONFIG));
         ClientMetrics.addTopologyDescriptionMetric(streamsMetrics, (metricsConfig, now) -> this.topologyMetadata.topologyDescriptionString());
         ClientMetrics.addStateMetric(streamsMetrics, (metricsConfig, now) -> state);
+        ClientMetrics.addClientStateTelemetryMetric(streamsMetrics, (metricsConfig, now) -> state.ordinal());
+        ClientMetrics.addClientRecordingLevelMetric(streamsMetrics, calculateMetricsRecordingLevel());
         threads = Collections.synchronizedList(new LinkedList<>());
         ClientMetrics.addNumAliveStreamThreadMetric(streamsMetrics, (metricsConfig, now) -> numLiveStreamThreads());
 
@@ -1067,7 +1070,7 @@ private StreamThread createAndAddStreamThread(final long cacheSizePerThread, fin
     private static Metrics createMetrics(final StreamsConfig config, final Time time, final String clientId) {
         final MetricConfig metricConfig = new MetricConfig()
             .samples(config.getInt(StreamsConfig.METRICS_NUM_SAMPLES_CONFIG))
-            .recordLevel(Sensor.RecordingLevel.forName(config.getString(StreamsConfig.METRICS_RECORDING_LEVEL_CONFIG)))
+            .recordLevel(Sensor.RecordingLevel.forName(config.getString(METRICS_RECORDING_LEVEL_CONFIG)))
             .timeWindow(config.getLong(StreamsConfig.METRICS_SAMPLE_WINDOW_MS_CONFIG), TimeUnit.MILLISECONDS);
         final List<MetricsReporter> reporters = CommonClientConfigs.metricsReporters(clientId, config);
 
@@ -1250,6 +1253,20 @@ private Optional<String> removeStreamThread(final long timeoutMs) throws Timeout
         return Optional.empty();
     }
 
+    private int calculateMetricsRecordingLevel() {
+        final int recordingLevel;
+        final String recordingLevelString = applicationConfigs.getString(METRICS_RECORDING_LEVEL_CONFIG);
+        if (recordingLevelString.equals("INFO")) {
+            recordingLevel = 0;
+        } else if (recordingLevelString.equals("DEBUG")) {
+            recordingLevel = 1;
+        } else {
+            // Must be TRACE level
+            recordingLevel = 2;
+        }
+        return recordingLevel;
+    }
+
     /*
      * Takes a snapshot and counts the number of stream threads which are not in PENDING_SHUTDOWN or DEAD
      *
@@ -1334,7 +1351,7 @@ private ScheduledExecutorService setupStateDirCleaner() {
 
     private static ScheduledExecutorService maybeCreateRocksDBMetricsRecordingService(final String clientId,
                                                                                       final StreamsConfig config) {
-        if (RecordingLevel.forName(config.getString(StreamsConfig.METRICS_RECORDING_LEVEL_CONFIG)) == RecordingLevel.DEBUG) {
+        if (RecordingLevel.forName(config.getString(METRICS_RECORDING_LEVEL_CONFIG)) == RecordingLevel.DEBUG) {
             return Executors.newSingleThreadScheduledExecutor(r -> {
                 final Thread thread = new Thread(r, clientId + "-RocksDBMetricsRecordingTrigger");
                 thread.setDaemon(true);

streams/src/main/java/org/apache/kafka/streams/internals/metrics/ClientMetrics.java

@@ -40,11 +40,14 @@ private ClientMetrics() {}
     private static final String APPLICATION_ID = "application-id";
     private static final String TOPOLOGY_DESCRIPTION = "topology-description";
     private static final String STATE = "state";
+    private static final String CLIENT_STATE = "client-state";
     private static final String ALIVE_STREAM_THREADS = "alive-stream-threads";
     private static final String VERSION_FROM_FILE;
     private static final String COMMIT_ID_FROM_FILE;
     private static final String DEFAULT_VALUE = "unknown";
     private static final String FAILED_STREAM_THREADS = "failed-stream-threads";
+    private static final String RECORDING_LEVEL = "recording-level";
+
 
     static {
         final Properties props = new Properties();
@@ -67,6 +70,7 @@ private ClientMetrics() {}
     private static final String STATE_DESCRIPTION = "The state of the Kafka Streams client";
     private static final String ALIVE_STREAM_THREADS_DESCRIPTION = "The current number of alive stream threads that are running or participating in rebalance";
     private static final String FAILED_STREAM_THREADS_DESCRIPTION = "The number of failed stream threads since the start of the Kafka Streams client";
+    private static final String RECORDING_LEVEL_DESCRIPTION = "The metrics recording level of the Kafka Streams client";
 
     public static String version() {
         return VERSION_FROM_FILE;
@@ -123,6 +127,26 @@ public static void addStateMetric(final StreamsMetricsImpl streamsMetrics,
         );
     }
 
+    public static void addClientStateTelemetryMetric(final StreamsMetricsImpl streamsMetrics,
+                                                     final Gauge<Integer> stateProvider) {
+        streamsMetrics.addClientLevelMutableMetric(
+            CLIENT_STATE,
+            STATE_DESCRIPTION,
+            RecordingLevel.INFO,
+            stateProvider
+        );
+    }
+
+    public static void addClientRecordingLevelMetric(final StreamsMetricsImpl streamsMetrics,
+                                                     final int recordingLevel) {
+        streamsMetrics.addClientLevelImmutableMetric(
+                RECORDING_LEVEL,
+                RECORDING_LEVEL_DESCRIPTION,
+                RecordingLevel.INFO,
+                recordingLevel
+        );
+    }
+
     public static void addNumAliveStreamThreadMetric(final StreamsMetricsImpl streamsMetrics,
                                                      final Gauge<Integer> stateProvider) {
         streamsMetrics.addClientLevelMutableMetric(

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

@@ -68,6 +68,7 @@
 import java.util.Comparator;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Queue;
@@ -614,6 +615,12 @@ public StreamThread(final Time time,
             streamsMetrics,
             time.milliseconds()
         );
+        ThreadMetrics.addThreadStateTelemetryMetric(threadId,
+                streamsMetrics,
+                (metricConfig, now) -> this.state().ordinal());
+        ThreadMetrics.addThreadStateMetric(threadId,
+                streamsMetrics,
+                (metricConfig, now) -> this.state().name().toLowerCase(Locale.getDefault()));
         ThreadMetrics.addThreadBlockedTimeMetric(
             threadId,
             new StreamThreadTotalBlockedTime(

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

@@ -16,6 +16,7 @@
  */
 package org.apache.kafka.streams.processor.internals.metrics;
 
+import org.apache.kafka.common.metrics.Gauge;
 import org.apache.kafka.common.metrics.Sensor;
 import org.apache.kafka.common.metrics.Sensor.RecordingLevel;
 import org.apache.kafka.streams.processor.internals.StreamThreadTotalBlockedTime;
@@ -44,7 +45,9 @@ private ThreadMetrics() {}
     private static final String CREATE_TASK = "task-created";
     private static final String CLOSE_TASK = "task-closed";
     private static final String BLOCKED_TIME = "blocked-time-ns-total";
+    private static final String STATE  = "state";
     private static final String THREAD_START_TIME = "thread-start-time";
+    private static final String THREAD_STATE = "thread-state";
 
     private static final String COMMIT_DESCRIPTION = "calls to commit";
     private static final String COMMIT_TOTAL_DESCRIPTION = TOTAL_DESCRIPTION + COMMIT_DESCRIPTION;
@@ -88,6 +91,8 @@ private ThreadMetrics() {}
         "The total time the thread spent blocked on kafka in nanoseconds";
     private static final String THREAD_START_TIME_DESCRIPTION =
         "The time that the thread was started";
+    private static final String THREAD_STATE_DESCRIPTION =
+        "The current state of the thread";
 
     public static Sensor createTaskSensor(final String threadId,
                                           final StreamsMetricsImpl streamsMetrics) {
@@ -290,6 +295,30 @@ public static void addThreadStartTimeMetric(final String threadId,
         );
     }
 
+    public static void addThreadStateTelemetryMetric(final String threadId,
+                                                     final StreamsMetricsImpl streamsMetrics,
+                                                     final Gauge<Integer> threadStateProvider) {
+        streamsMetrics.addThreadLevelMutableMetric(
+                THREAD_STATE,
+                THREAD_STATE_DESCRIPTION,
+                threadId,
+                threadStateProvider
+        );
+    }
+
+    public static void addThreadStateMetric(final String threadId,
+                                            final StreamsMetricsImpl streamsMetrics,
+                                            final Gauge<String> threadStateProvider) {
+        streamsMetrics.addThreadLevelMutableMetric(
+                STATE,
+                THREAD_STATE_DESCRIPTION,
+                threadId,
+                threadStateProvider
+        );
+    }
+
+
+
     public static void addThreadBlockedTimeMetric(final String threadId,
                                                   final StreamThreadTotalBlockedTime blockedTime,
                                                   final StreamsMetricsImpl streamsMetrics) {

streams/src/test/java/org/apache/kafka/streams/internals/metrics/ClientMetricsTest.java

@@ -111,6 +111,32 @@ public void shouldAddAliveStreamThreadsMetric() {
         );
     }
 
+    @Test
+    public void shouldAddClientStateTelemetryMetric() {
+        final String name = "client-state";
+        final String description = "The state of the Kafka Streams client";
+        final Gauge<Integer> stateProvider = (config, now) -> State.RUNNING.ordinal();
+        setUpAndVerifyMutableMetric(
+                name,
+                description,
+                stateProvider,
+                () -> ClientMetrics.addClientStateTelemetryMetric(streamsMetrics, stateProvider)
+        );
+    }
+
+    @Test
+    public void shouldAddRecordingLevelMetric() {
+        final String name = "recording-level";
+        final String description = "The metrics recording level of the Kafka Streams client";
+        final int recordingLevel = 1;
+        setUpAndVerifyImmutableMetric(
+                name,
+                description,
+                recordingLevel,
+                () -> ClientMetrics.addClientRecordingLevelMetric(streamsMetrics, recordingLevel)
+        );
+    }
+
     @Test
     public void shouldGetFailedStreamThreadsSensor() {
         final String name = "failed-stream-threads";
@@ -159,4 +185,19 @@ private void setUpAndVerifyImmutableMetric(final String name,
                 eq(value)
         );
     }
+
+    private void setUpAndVerifyImmutableMetric(final String name,
+                                               final String description,
+                                               final int value,
+                                               final Runnable metricAdder) {
+
+        metricAdder.run();
+
+        verify(streamsMetrics).addClientLevelImmutableMetric(
+                eq(name),
+                eq(description),
+                eq(RecordingLevel.INFO),
+                eq(value)
+        );
+    }
 }

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

@@ -19,13 +19,15 @@
 import org.apache.kafka.common.metrics.Gauge;
 import org.apache.kafka.common.metrics.Sensor;
 import org.apache.kafka.common.metrics.Sensor.RecordingLevel;
+import org.apache.kafka.streams.processor.internals.StreamThread;
 import org.apache.kafka.streams.processor.internals.StreamThreadTotalBlockedTime;
 
 import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentCaptor;
 import org.mockito.MockedStatic;
 
 import java.util.Collections;
+import java.util.Locale;
 import java.util.Map;
 
 import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.LATENCY_SUFFIX;
@@ -413,6 +415,39 @@ public void shouldAddThreadStartTimeMetric() {
         );
     }
 
+    @Test
+    public void shouldAddThreadStateTelemetryMetric() {
+        final Gauge<Integer> threadStateProvider = (streamsMetrics, startTime) -> StreamThread.State.RUNNING.ordinal();
+        ThreadMetrics.addThreadStateTelemetryMetric(
+                "threadId",
+                streamsMetrics,
+                threadStateProvider
+        );
+        verify(streamsMetrics).addThreadLevelMutableMetric(
+                "thread-state",
+                "The current state of the thread",
+                "threadId",
+                threadStateProvider
+        );
+    }
+
+    @Test
+    public void shouldAddThreadStateJMXMetric() {
+        final Gauge<String> threadStateProvider = (streamsMetrics, startTime) -> StreamThread.State.RUNNING.name().toLowerCase(Locale.getDefault());
+        ThreadMetrics.addThreadStateMetric(
+                "threadId",
+                streamsMetrics,
+                threadStateProvider
+        );
+        verify(streamsMetrics).addThreadLevelMutableMetric(
+                "state",
+                "The current state of the thread",
+                "threadId",
+                threadStateProvider
+        );
+    }
+
+
     @Test
     public void shouldAddTotalBlockedTimeMetric() {
         // Given: