KAFKA-8934: Introduce instance-level metrics for streams applications

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

@@ -37,6 +37,7 @@
 import org.apache.kafka.streams.errors.ProcessorStateException;
 import org.apache.kafka.streams.errors.StreamsException;
 import org.apache.kafka.streams.internals.ApiUtils;
+import org.apache.kafka.streams.internals.metrics.ClientMetrics;
 import org.apache.kafka.streams.kstream.KStream;
 import org.apache.kafka.streams.kstream.KTable;
 import org.apache.kafka.streams.kstream.Produced;
@@ -53,6 +54,7 @@
 import org.apache.kafka.streams.processor.internals.StreamThread;
 import org.apache.kafka.streams.processor.internals.StreamsMetadataState;
 import org.apache.kafka.streams.processor.internals.ThreadStateTransitionValidator;
+import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
 import org.apache.kafka.streams.state.HostInfo;
 import org.apache.kafka.streams.state.QueryableStoreType;
 import org.apache.kafka.streams.state.StreamsMetadata;
@@ -144,6 +146,7 @@ public class KafkaStreams implements AutoCloseable {
     private final ScheduledExecutorService rocksDBMetricsRecordingService;
     private final QueryableStoreProvider queryableStoreProvider;
     private final Admin adminClient;
+    private final StreamsMetricsImpl streamsMetrics;
 
     GlobalStreamThread globalStreamThread;
     private KafkaStreams.StateListener stateListener;
@@ -672,6 +675,16 @@ private KafkaStreams(final InternalTopologyBuilder internalTopologyBuilder,
                 Collections.singletonMap(StreamsConfig.CLIENT_ID_CONFIG, clientId));
         reporters.add(new JmxReporter(JMX_PREFIX));
         metrics = new Metrics(metricConfig, reporters, time);
+        streamsMetrics =
+            new StreamsMetricsImpl(metrics, clientId, config.getString(StreamsConfig.BUILT_IN_METRICS_VERSION_CONFIG));
+        streamsMetrics.setRocksDBMetricsRecordingTrigger(rocksDBMetricsRecordingTrigger);
+        ClientMetrics.addVersionMetric(streamsMetrics);
+        ClientMetrics.addCommitIdMetric(streamsMetrics);
+        ClientMetrics.addApplicationIdMetric(streamsMetrics, config.getString(StreamsConfig.APPLICATION_ID_CONFIG));
+        ClientMetrics.addTopologyDescriptionMetric(streamsMetrics, internalTopologyBuilder.describe().toString());
+        ClientMetrics.addStateMetric(streamsMetrics, (metricsConfig, now) -> state);
+        log.info("Kafka Streams version: {}", ClientMetrics.version());
+        log.info("Kafka Streams commit ID: {}", ClientMetrics.commitId());
 
         // re-write the physical topology according to the config
         internalTopologyBuilder.rewriteTopology(config);
@@ -712,11 +725,10 @@ private KafkaStreams(final InternalTopologyBuilder internalTopologyBuilder,
                 clientSupplier.getGlobalConsumer(config.getGlobalConsumerConfigs(clientId)),
                 stateDirectory,
                 cacheSizePerThread,
-                metrics,
+                streamsMetrics,
                 time,
                 globalThreadId,
-                delegatingStateRestoreListener,
-                rocksDBMetricsRecordingTrigger
+                delegatingStateRestoreListener
             );
             globalThreadState = globalStreamThread.state();
         }
@@ -734,14 +746,13 @@ private KafkaStreams(final InternalTopologyBuilder internalTopologyBuilder,
                 adminClient,
                 processId,
                 clientId,
-                metrics,
+                streamsMetrics,
                 time,
                 streamsMetadataState,
                 cacheSizePerThread,
                 stateDirectory,
                 delegatingStateRestoreListener,
                 i + 1);
-            threads[i].setRocksDBMetricsRecordingTrigger(rocksDBMetricsRecordingTrigger);
             threadState.put(threads[i].getId(), threads[i].state());
             storeProviders.add(new StreamThreadStateStoreProvider(threads[i]));
         }
@@ -928,6 +939,7 @@ private boolean close(final long timeoutMs) {
 
                 adminClient.close();
 
+                streamsMetrics.removeAllClientLevelMetrics();
                 metrics.close();
                 setState(State.NOT_RUNNING);
             }, "kafka-streams-close-thread");

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

@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.internals.metrics;
+
+import org.apache.kafka.common.metrics.Gauge;
+import org.apache.kafka.common.metrics.Sensor.RecordingLevel;
+import org.apache.kafka.streams.KafkaStreams.State;
+import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.InputStream;
+import java.util.Properties;
+
+public class ClientMetrics {
+    private ClientMetrics() {}
+
+    private static final Logger log = LoggerFactory.getLogger(ClientMetrics.class);
+    private static final String VERSION = "version";
+    private static final String COMMIT_ID = "commit-id";
+    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 VERSION_FROM_FILE;
+    private static final String COMMIT_ID_FROM_FILE;
+    private static final String DEFAULT_VALUE = "unknown";
+
+    static {
+        final Properties props = new Properties();
+        try (InputStream resourceStream = ClientMetrics.class.getResourceAsStream(
+            "/kafka/kafka-streams-version.properties")) {
+
+            props.load(resourceStream);
+        } catch (final Exception exception) {
+            log.warn("Error while loading kafka-streams-version.properties", exception);
+        }
+        VERSION_FROM_FILE = props.getProperty("version", DEFAULT_VALUE).trim();
+        COMMIT_ID_FROM_FILE = props.getProperty("commitId", DEFAULT_VALUE).trim();
+    }
+
+    private static final String VERSION_DESCRIPTION = "The version of the Kafka Streams client";
+    private static final String COMMIT_ID_DESCRIPTION = "The version control commit ID of the Kafka Streams client";
+    private static final String APPLICATION_ID_DESCRIPTION = "The application ID of the Kafka Streams client";
+    private static final String TOPOLOGY_DESCRIPTION_DESCRIPTION =
+        "The description of the topology executed in the Kafka Streams client";
+    private static final String STATE_DESCRIPTION = "The state of the Kafka Streams client";
+
+    public static String version() {
+        return VERSION_FROM_FILE;
+    }
+
+    public static String commitId() {
+        return COMMIT_ID_FROM_FILE;
+    }
+
+    public static void addVersionMetric(final StreamsMetricsImpl streamsMetrics) {
+        streamsMetrics.addClientLevelImmutableMetric(
+            VERSION,
+            VERSION_DESCRIPTION,
+            RecordingLevel.INFO,
+            VERSION_FROM_FILE
+        );
+    }
+
+    public static void addCommitIdMetric(final StreamsMetricsImpl streamsMetrics) {
+        streamsMetrics.addClientLevelImmutableMetric(
+            COMMIT_ID,
+            COMMIT_ID_DESCRIPTION,
+            RecordingLevel.INFO,
+            COMMIT_ID_FROM_FILE
+        );
+    }
+
+    public static void addApplicationIdMetric(final StreamsMetricsImpl streamsMetrics, final String applicationId) {
+        streamsMetrics.addClientLevelImmutableMetric(
+            APPLICATION_ID,
+            APPLICATION_ID_DESCRIPTION,
+            RecordingLevel.INFO,
+            applicationId
+        );
+    }
+
+    public static void addTopologyDescriptionMetric(final StreamsMetricsImpl streamsMetrics,
+                                                    final String topologyDescription) {
+        streamsMetrics.addClientLevelImmutableMetric(
+            TOPOLOGY_DESCRIPTION,
+            TOPOLOGY_DESCRIPTION_DESCRIPTION,
+            RecordingLevel.INFO,
+            topologyDescription
+        );
+    }
+
+    public static void addStateMetric(final StreamsMetricsImpl streamsMetrics,
+                                      final Gauge<State> stateProvider) {
+        streamsMetrics.addClientLevelMutableMetric(
+            STATE,
+            STATE_DESCRIPTION,
+            RecordingLevel.INFO,
+            stateProvider
+        );
+    }
+}

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamAggregate.java

@@ -67,7 +67,7 @@ private class KStreamAggregateProcessor extends AbstractProcessor<K, V> {
         public void init(final ProcessorContext context) {
             super.init(context);
             metrics = (StreamsMetricsImpl) context.metrics();
-            skippedRecordsSensor = ThreadMetrics.skipRecordSensor(metrics);
+            skippedRecordsSensor = ThreadMetrics.skipRecordSensor(Thread.currentThread().getName(), metrics);
             store = (TimestampedKeyValueStore<K, T>) context.getStateStore(storeName);
             tupleForwarder = new TimestampedTupleForwarder<>(
                 store,

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoin.java

@@ -65,7 +65,7 @@ private class KStreamKStreamJoinProcessor extends AbstractProcessor<K, V1> {
         public void init(final ProcessorContext context) {
             super.init(context);
             metrics = (StreamsMetricsImpl) context.metrics();
-            skippedRecordsSensor = ThreadMetrics.skipRecordSensor(metrics);
+            skippedRecordsSensor = ThreadMetrics.skipRecordSensor(Thread.currentThread().getName(), metrics);
 
             otherWindow = (WindowStore<K, V2>) context.getStateStore(otherWindowName);
         }

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinProcessor.java

@@ -52,7 +52,7 @@ class KStreamKTableJoinProcessor<K1, K2, V1, V2, R> extends AbstractProcessor<K1
     public void init(final ProcessorContext context) {
         super.init(context);
         metrics = (StreamsMetricsImpl) context.metrics();
-        skippedRecordsSensor = ThreadMetrics.skipRecordSensor(metrics);
+        skippedRecordsSensor = ThreadMetrics.skipRecordSensor(Thread.currentThread().getName(), metrics);
 
         valueGetter.init(context);
     }

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamReduce.java

@@ -65,7 +65,7 @@ private class KStreamReduceProcessor extends AbstractProcessor<K, V> {
         public void init(final ProcessorContext context) {
             super.init(context);
             metrics = (StreamsMetricsImpl) context.metrics();
-            skippedRecordsSensor = ThreadMetrics.skipRecordSensor(metrics);
+            skippedRecordsSensor = ThreadMetrics.skipRecordSensor(Thread.currentThread().getName(), metrics);
             store = (TimestampedKeyValueStore<K, V>) context.getStateStore(storeName);
             tupleForwarder = new TimestampedTupleForwarder<>(
                 store,

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregate.java

@@ -94,7 +94,7 @@ public void init(final ProcessorContext context) {
             internalProcessorContext = (InternalProcessorContext) context;
             metrics = (StreamsMetricsImpl) context.metrics();
             lateRecordDropSensor = Sensors.lateRecordDropSensor(internalProcessorContext);
-            skippedRecordsSensor = ThreadMetrics.skipRecordSensor(metrics);
+            skippedRecordsSensor = ThreadMetrics.skipRecordSensor(Thread.currentThread().getName(), metrics);
 
             store = (SessionStore<K, Agg>) context.getStateStore(storeName);
             tupleForwarder = new SessionTupleForwarder<>(store, context, new SessionCacheFlushListener<>(context), sendOldValues);

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamWindowAggregate.java

@@ -92,7 +92,7 @@ public void init(final ProcessorContext context) {
             metrics = internalProcessorContext.metrics();
 
             lateRecordDropSensor = Sensors.lateRecordDropSensor(internalProcessorContext);
-            skippedRecordsSensor = ThreadMetrics.skipRecordSensor(metrics);
+            skippedRecordsSensor = ThreadMetrics.skipRecordSensor(Thread.currentThread().getName(), metrics);
             windowStore = (TimestampedWindowStore<K, Agg>) context.getStateStore(storeName);
             tupleForwarder = new TimestampedTupleForwarder<>(
                 windowStore,

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableInnerJoin.java

@@ -78,7 +78,7 @@ private class KTableKTableJoinProcessor extends AbstractProcessor<K, Change<V1>>
         public void init(final ProcessorContext context) {
             super.init(context);
             metrics = (StreamsMetricsImpl) context.metrics();
-            skippedRecordsSensor = ThreadMetrics.skipRecordSensor(metrics);
+            skippedRecordsSensor = ThreadMetrics.skipRecordSensor(Thread.currentThread().getName(), metrics);
             valueGetter.init(context);
         }
 

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableLeftJoin.java

@@ -77,7 +77,7 @@ private class KTableKTableLeftJoinProcessor extends AbstractProcessor<K, Change<
         public void init(final ProcessorContext context) {
             super.init(context);
             metrics = (StreamsMetricsImpl) context.metrics();
-            skippedRecordsSensor = ThreadMetrics.skipRecordSensor(metrics);
+            skippedRecordsSensor = ThreadMetrics.skipRecordSensor(Thread.currentThread().getName(), metrics);
             valueGetter.init(context);
         }
 

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableOuterJoin.java

@@ -76,7 +76,7 @@ private class KTableKTableOuterJoinProcessor extends AbstractProcessor<K, Change
         public void init(final ProcessorContext context) {
             super.init(context);
             metrics = (StreamsMetricsImpl) context.metrics();
-            skippedRecordsSensor = ThreadMetrics.skipRecordSensor(metrics);
+            skippedRecordsSensor = ThreadMetrics.skipRecordSensor(Thread.currentThread().getName(), metrics);
             valueGetter.init(context);
         }
 

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableRightJoin.java

@@ -75,7 +75,7 @@ private class KTableKTableRightJoinProcessor extends AbstractProcessor<K, Change
         public void init(final ProcessorContext context) {
             super.init(context);
             metrics = (StreamsMetricsImpl) context.metrics();
-            skippedRecordsSensor = ThreadMetrics.skipRecordSensor(metrics);
+            skippedRecordsSensor = ThreadMetrics.skipRecordSensor(Thread.currentThread().getName(), metrics);
             valueGetter.init(context);
         }
 

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableSource.java

@@ -79,7 +79,7 @@ private class KTableSourceProcessor extends AbstractProcessor<K, V> {
         public void init(final ProcessorContext context) {
             super.init(context);
             metrics = (StreamsMetricsImpl) context.metrics();
-            skippedRecordsSensor = ThreadMetrics.skipRecordSensor(metrics);
+            skippedRecordsSensor = ThreadMetrics.skipRecordSensor(Thread.currentThread().getName(), metrics);
             if (queryableName != null) {
                 store = (TimestampedKeyValueStore<K, V>) context.getStateStore(queryableName);
                 tupleForwarder = new TimestampedTupleForwarder<>(

streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/ForeignJoinSubscriptionProcessorSupplier.java

@@ -64,7 +64,8 @@ private final class KTableKTableJoinProcessor extends AbstractProcessor<KO, Chan
         public void init(final ProcessorContext context) {
             super.init(context);
             final InternalProcessorContext internalProcessorContext = (InternalProcessorContext) context;
-            skippedRecordsSensor = ThreadMetrics.skipRecordSensor(internalProcessorContext.metrics());
+            skippedRecordsSensor =
+                ThreadMetrics.skipRecordSensor(Thread.currentThread().getName(), internalProcessorContext.metrics());
             store = internalProcessorContext.getStateStore(storeBuilder);
         }
 

streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionStoreReceiveProcessorSupplier.java

@@ -65,7 +65,7 @@ public void init(final ProcessorContext context) {
                 final InternalProcessorContext internalProcessorContext = (InternalProcessorContext) context;
 
                 metrics = internalProcessorContext.metrics();
-                skippedRecordsSensor = ThreadMetrics.skipRecordSensor(metrics);
+                skippedRecordsSensor = ThreadMetrics.skipRecordSensor(Thread.currentThread().getName(), metrics);
                 store = internalProcessorContext.getStateStore(storeBuilder);
             }
 

streams/src/main/java/org/apache/kafka/streams/kstream/internals/metrics/Sensors.java

@@ -38,7 +38,10 @@ private Sensors() {}
 
     public static Sensor lateRecordDropSensor(final InternalProcessorContext context) {
         final StreamsMetricsImpl metrics = context.metrics();
+        final String threadId = Thread.currentThread().getName();
+
         final Sensor sensor = metrics.nodeLevelSensor(
+            threadId,
             context.taskId().toString(),
             context.currentNode().name(),
             LATE_RECORD_DROP,
@@ -47,23 +50,33 @@ public static Sensor lateRecordDropSensor(final InternalProcessorContext context
         StreamsMetricsImpl.addInvocationRateAndCountToSensor(
             sensor,
             PROCESSOR_NODE_METRICS_GROUP,
-            metrics.tagMap("task-id", context.taskId().toString(), PROCESSOR_NODE_ID_TAG, context.currentNode().name()),
+            metrics.tagMap(
+                threadId,
+                "task-id",
+                context.taskId().toString(),
+                PROCESSOR_NODE_ID_TAG,
+                context.currentNode().name()
+            ),
             LATE_RECORD_DROP
         );
         return sensor;
     }
 
     public static Sensor recordLatenessSensor(final InternalProcessorContext context) {
         final StreamsMetricsImpl metrics = context.metrics();
+        final String threadId = Thread.currentThread().getName();
 
         final Sensor sensor = metrics.taskLevelSensor(
+            threadId,
             context.taskId().toString(),
             "record-lateness",
             Sensor.RecordingLevel.DEBUG
         );
 
         final Map<String, String> tags = metrics.tagMap(
-            "task-id", context.taskId().toString()
+            threadId,
+            "task-id",
+            context.taskId().toString()
         );
         sensor.add(
             new MetricName(
@@ -86,17 +99,22 @@ public static Sensor recordLatenessSensor(final InternalProcessorContext context
 
     public static Sensor suppressionEmitSensor(final InternalProcessorContext context) {
         final StreamsMetricsImpl metrics = context.metrics();
+        final String threadId = Thread.currentThread().getName();
 
         final Sensor sensor = metrics.nodeLevelSensor(
+            threadId,
             context.taskId().toString(),
             context.currentNode().name(),
             "suppression-emit",
             Sensor.RecordingLevel.DEBUG
         );
 
         final Map<String, String> tags = metrics.tagMap(
-            "task-id", context.taskId().toString(),
-            PROCESSOR_NODE_ID_TAG, context.currentNode().name()
+            threadId,
+            "task-id",
+            context.taskId().toString(),
+            PROCESSOR_NODE_ID_TAG,
+            context.currentNode().name()
         );
 
         sensor.add(

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

@@ -70,7 +70,7 @@ public Map<TopicPartition, Long> initialize() {
                     source,
                     deserializationExceptionHandler,
                     logContext,
-                    ThreadMetrics.skipRecordSensor(processorContext.metrics())
+                    ThreadMetrics.skipRecordSensor(Thread.currentThread().getName(), processorContext.metrics())
                 )
             );
         }