KAFKA-19734 Adding application tag to clientstate metric (#20766)

## Summary

As a follow-on to the improvements introduced in

[KIP-1091](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1091%3A+Improved+Kafka+Streams+operator+metrics)
it would be useful to add the application-id as a tag to the
`client-state` metric. This allows Kafka Streams developers and
operators to connect metrics containing a `thread-id` (which embeds the
`application-id`) across separate deployments of Kafka Streams
instances, which are members of the same logical application.

Reviewers: Bill Bejeck<[email protected]> Matthias
 Sax<[email protected]>

Author: gensericghiro

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

@@ -983,6 +983,7 @@ private KafkaStreams(final TopologyMetadata topologyMetadata,
             metrics,
             clientId,
             processId.toString(),
+            applicationId,
             time
         );
 

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

@@ -90,6 +90,7 @@ public int hashCode() {
     private final Map<Sensor, Sensor> parentSensors;
     private final String clientId;
     private final String processId;
+    private final String applicationId;
 
     private final Version version;
     private final Deque<MetricName> clientLevelMetrics = new LinkedList<>();
@@ -118,6 +119,7 @@ public int hashCode() {
 
     public static final String CLIENT_ID_TAG = "client-id";
     public static final String PROCESS_ID_TAG = "process-id";
+    public static final String APPLICATION_ID_TAG = "application-id";
     public static final String THREAD_ID_TAG = "thread-id";
     public static final String TASK_ID_TAG = "task-id";
     public static final String PROCESSOR_NODE_ID_TAG = "processor-node-id";
@@ -168,11 +170,13 @@ public int hashCode() {
     public StreamsMetricsImpl(final Metrics metrics,
                               final String clientId,
                               final String processId,
+                              final String applicationId,
                               final Time time) {
         Objects.requireNonNull(metrics, "Metrics cannot be null");
         this.metrics = metrics;
         this.clientId = clientId;
         this.processId = processId;
+        this.applicationId = applicationId;
         version = Version.LATEST;
         rocksDBMetricsRecordingTrigger = new RocksDBMetricsRecordingTrigger(time);
 
@@ -284,6 +288,7 @@ public Map<String, String> clientLevelTagMap() {
         final Map<String, String> tagMap = new LinkedHashMap<>();
         tagMap.put(CLIENT_ID_TAG, clientId);
         tagMap.put(PROCESS_ID_TAG, processId);
+        tagMap.put(APPLICATION_ID_TAG, applicationId);
         return tagMap;
     }
 

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregateProcessorTest.java

@@ -80,7 +80,7 @@ public class KStreamSessionWindowAggregateProcessorTest {
 
     private final MockTime time = new MockTime();
     private final Metrics metrics = new Metrics();
-    private final StreamsMetricsImpl streamsMetrics = new StreamsMetricsImpl(metrics, "test", "processId", time);
+    private final StreamsMetricsImpl streamsMetrics = new StreamsMetricsImpl(metrics, "test", "processId", "applicationId", time);
     private final String threadId = Thread.currentThread().getName();
     private final Initializer<Long> initializer = () -> 0L;
     private final Aggregator<String, String, Long> aggregator = (aggKey, value, aggregate) -> aggregate + 1;

streams/src/test/java/org/apache/kafka/streams/processor/internals/ActiveTaskCreatorTest.java

@@ -72,7 +72,7 @@ public class ActiveTaskCreatorTest {
     private ChangelogReader changeLogReader;
 
     private final MockClientSupplier mockClientSupplier = new MockClientSupplier();
-    private final StreamsMetricsImpl streamsMetrics = new StreamsMetricsImpl(new Metrics(), "clientId", "processId", new MockTime());
+    private final StreamsMetricsImpl streamsMetrics = new StreamsMetricsImpl(new Metrics(), "clientId", "processId", "applicationId", new MockTime());
     private final Map<String, Object> properties = mkMap(
         mkEntry(StreamsConfig.APPLICATION_ID_CONFIG, "appId"),
         mkEntry(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:1234")

streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdaterTest.java

@@ -106,7 +106,7 @@ class DefaultStateUpdaterTest {
 
     // need an auto-tick timer to work for draining with timeout
     private final Time time = new MockTime(1L);
-    private final StreamsMetricsImpl metrics = new StreamsMetricsImpl(new Metrics(time), "", "", time);
+    private final StreamsMetricsImpl metrics = new StreamsMetricsImpl(new Metrics(time), "", "", "", time);
     private final StreamsConfig config = new StreamsConfig(configProps(COMMIT_INTERVAL));
     private final ChangelogReader changelogReader = mock(ChangelogReader.class);
     private final TopologyMetadata topologyMetadata = unnamedTopology().build();

streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStreamThreadTest.java

@@ -132,7 +132,7 @@ public void process(final Record<Object, Object> record) {
             mockConsumer,
             new StateDirectory(config, time, true, false),
             0,
-            new StreamsMetricsImpl(new Metrics(), "test-client", "processId", time),
+            new StreamsMetricsImpl(new Metrics(), "test-client", "processId", "applicationId", time),
             time,
             "clientId",
             stateRestoreListener,
@@ -169,7 +169,7 @@ public List<PartitionInfo> partitionsFor(final String topic) {
             mockConsumer,
             new StateDirectory(config, time, true, false),
             0,
-            new StreamsMetricsImpl(new Metrics(), "test-client", "processId", time),
+            new StreamsMetricsImpl(new Metrics(), "test-client", "processId", "applicationId", time),
             time,
             "clientId",
             stateRestoreListener,
@@ -418,7 +418,7 @@ public List<PartitionInfo> partitionsFor(final String topic) {
                 consumer,
                 new StateDirectory(config, time, true, false),
                 0,
-                new StreamsMetricsImpl(new Metrics(), "test-client", "processId", time),
+                new StreamsMetricsImpl(new Metrics(), "test-client", "processId", "applicationId", time),
                 time,
                 "clientId",
                 stateRestoreListener,

streams/src/test/java/org/apache/kafka/streams/processor/internals/MockStreamsMetrics.java

@@ -23,6 +23,6 @@
 public class MockStreamsMetrics extends StreamsMetricsImpl {
 
     public MockStreamsMetrics(final Metrics metrics) {
-        super(metrics, "test", "processId", new MockTime());
+        super(metrics, "test", "processId", "applicationId", new MockTime());
     }
 }

streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorNodeTest.java

@@ -279,7 +279,7 @@ public Set<TopicPartition> partitions() {
     public void testMetricsWithBuiltInMetricsVersionLatest() {
         final Metrics metrics = new Metrics();
         final StreamsMetricsImpl streamsMetrics =
-            new StreamsMetricsImpl(metrics, "test-client", "processId", new MockTime());
+            new StreamsMetricsImpl(metrics, "test-client", "processId", "applicationId", new MockTime());
         final InternalMockProcessorContext<Object, Object> context = new InternalMockProcessorContext<>(streamsMetrics);
         final ProcessorNode<Object, Object, Object, Object> node =
             new ProcessorNode<>(NAME, new NoOpProcessor(), Collections.emptySet());
@@ -363,7 +363,7 @@ public void process(final Record<Object, Object> record) {
     public void testTopologyLevelClassCastExceptionDirect() {
         final Metrics metrics = new Metrics();
         final StreamsMetricsImpl streamsMetrics =
-            new StreamsMetricsImpl(metrics, "test-client", "processId", new MockTime());
+            new StreamsMetricsImpl(metrics, "test-client", "processId", "applicationId", new MockTime());
         final InternalMockProcessorContext<Object, Object> context = new InternalMockProcessorContext<>(streamsMetrics);
         final ProcessorNode<Object, Object, Object, Object> node =
             new ProcessorNode<>("pname", new ClassCastProcessor(), Collections.emptySet());
@@ -441,7 +441,7 @@ private InternalProcessorContext<Object, Object> mockInternalProcessorContext()
         final InternalProcessorContext<Object, Object> internalProcessorContext = mock(InternalProcessorContext.class, withSettings().strictness(Strictness.LENIENT));
 
         when(internalProcessorContext.taskId()).thenReturn(TASK_ID);
-        when(internalProcessorContext.metrics()).thenReturn(new StreamsMetricsImpl(new Metrics(), "test-client", "processId", new MockTime()));
+        when(internalProcessorContext.metrics()).thenReturn(new StreamsMetricsImpl(new Metrics(), "test-client", "processId", "applicationId", new MockTime()));
         when(internalProcessorContext.topic()).thenReturn(TOPIC);
         when(internalProcessorContext.partition()).thenReturn(PARTITION);
         when(internalProcessorContext.offset()).thenReturn(OFFSET);

streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordQueueTest.java

@@ -72,7 +72,7 @@ public class RecordQueueTest {
 
     private final Metrics metrics = new Metrics();
     private final StreamsMetricsImpl streamsMetrics =
-        new StreamsMetricsImpl(metrics, "mock", "processId", new MockTime());
+        new StreamsMetricsImpl(metrics, "mock", "processId", "applicationId", new MockTime());
 
     final InternalMockProcessorContext<Integer, Integer> context = new InternalMockProcessorContext<>(
         StateSerdes.withBuiltinTypes("anyName", Bytes.class, Bytes.class),

streams/src/test/java/org/apache/kafka/streams/processor/internals/SourceNodeTest.java

@@ -97,7 +97,7 @@ public String deserialize(final String topic, final byte[] data) {
     public void shouldExposeProcessMetrics() {
         final Metrics metrics = new Metrics();
         final StreamsMetricsImpl streamsMetrics =
-            new StreamsMetricsImpl(metrics, "test-client", "processId", new MockTime());
+            new StreamsMetricsImpl(metrics, "test-client", "processId", "applicationId", new MockTime());
         final InternalMockProcessorContext<String, String> context = new InternalMockProcessorContext<>(streamsMetrics);
         final SourceNode<String, String> node =
             new SourceNode<>(context.currentNode().name(), new TheDeserializer(), new TheDeserializer());