KAFKA-19882: Removing process id from default client level tags (#20939)

## Summary
As a follow-up to #20906, also removing the process-id tag from 4.1.
This is because when working on
KIP-1091, we mistakenly applied the process-id tag to all client-level
metrics, rather than just the client-state, thread-state, and
recording-level metrics as specified in the KIP.

## Tests
Unit tests in `ClientMetricsTest.java` and `StreamsMetricsImplTest.java`

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

Author: gensericghiro

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

@@ -980,7 +980,6 @@ private KafkaStreams(final TopologyMetadata topologyMetadata,
         streamsMetrics = new StreamsMetricsImpl(
             metrics,
             clientId,
-            processId.toString(),
             time
         );
 
@@ -989,8 +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.name());
-        ClientMetrics.addClientStateTelemetryMetric(streamsMetrics, (metricsConfig, now) -> state.ordinal());
-        ClientMetrics.addClientRecordingLevelMetric(streamsMetrics, calculateMetricsRecordingLevel());
+        ClientMetrics.addClientStateTelemetryMetric(processId.toString(), streamsMetrics, (metricsConfig, now) -> state.ordinal());
+        ClientMetrics.addClientRecordingLevelMetric(processId.toString(), streamsMetrics, calculateMetricsRecordingLevel());
         threads = Collections.synchronizedList(new LinkedList<>());
         ClientMetrics.addNumAliveStreamThreadMetric(streamsMetrics, (metricsConfig, now) -> numLiveStreamThreads());
 

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

@@ -25,9 +25,11 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.InputStream;
+import java.util.Collections;
 import java.util.Properties;
 
 import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.CLIENT_LEVEL_GROUP;
+import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.PROCESS_ID_TAG;
 import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addSumMetricToSensor;
 
 public class ClientMetrics {
@@ -126,21 +128,25 @@ public static void addStateMetric(final StreamsMetricsImpl streamsMetrics,
         );
     }
 
-    public static void addClientStateTelemetryMetric(final StreamsMetricsImpl streamsMetrics,
+    public static void addClientStateTelemetryMetric(final String processId,
+                                                     final StreamsMetricsImpl streamsMetrics,
                                                      final Gauge<Integer> stateProvider) {
         streamsMetrics.addClientLevelMutableMetric(
             CLIENT_STATE,
             STATE_DESCRIPTION,
+            Collections.singletonMap(PROCESS_ID_TAG, processId),
             RecordingLevel.INFO,
             stateProvider
         );
     }
 
-    public static void addClientRecordingLevelMetric(final StreamsMetricsImpl streamsMetrics,
+    public static void addClientRecordingLevelMetric(final String processId,
+                                                     final StreamsMetricsImpl streamsMetrics,
                                                      final int recordingLevel) {
         streamsMetrics.addClientLevelImmutableMetric(
                 RECORDING_LEVEL,
                 RECORDING_LEVEL_DESCRIPTION,
+                Collections.singletonMap(PROCESS_ID_TAG, processId),
                 RecordingLevel.INFO,
                 recordingLevel
         );

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

@@ -89,7 +89,6 @@ public int hashCode() {
     private final Metrics metrics;
     private final Map<Sensor, Sensor> parentSensors;
     private final String clientId;
-    private final String processId;
 
     private final Version version;
     private final Deque<MetricName> clientLevelMetrics = new LinkedList<>();
@@ -167,12 +166,10 @@ public int hashCode() {
 
     public StreamsMetricsImpl(final Metrics metrics,
                               final String clientId,
-                              final String processId,
                               final Time time) {
         Objects.requireNonNull(metrics, "Metrics cannot be null");
         this.metrics = metrics;
         this.clientId = clientId;
-        this.processId = processId;
         version = Version.LATEST;
         rocksDBMetricsRecordingTrigger = new RocksDBMetricsRecordingTrigger(time);
 
@@ -195,7 +192,20 @@ public <T> void addClientLevelImmutableMetric(final String name,
                                                   final String description,
                                                   final RecordingLevel recordingLevel,
                                                   final T value) {
-        final MetricName metricName = metrics.metricName(name, CLIENT_LEVEL_GROUP, description, clientLevelTagMap());
+        addClientLevelImmutableMetric(name, description, Collections.emptyMap(), recordingLevel, value);
+    }
+
+    public <T> void addClientLevelImmutableMetric(final String name,
+                                                  final String description,
+                                                  final Map<String, String> additionalTags,
+                                                  final RecordingLevel recordingLevel,
+                                                  final T value) {
+        final MetricName metricName = metrics.metricName(
+            name,
+            CLIENT_LEVEL_GROUP,
+            description,
+            clientLevelTagMap(additionalTags)
+        );
         final MetricConfig metricConfig = new MetricConfig().recordLevel(recordingLevel);
         synchronized (clientLevelMetrics) {
             metrics.addMetric(metricName, metricConfig, new ImmutableMetricValue<>(value));
@@ -207,7 +217,20 @@ public <T> void addClientLevelMutableMetric(final String name,
                                                 final String description,
                                                 final RecordingLevel recordingLevel,
                                                 final Gauge<T> valueProvider) {
-        final MetricName metricName = metrics.metricName(name, CLIENT_LEVEL_GROUP, description, clientLevelTagMap());
+        addClientLevelMutableMetric(name, description, Collections.emptyMap(), recordingLevel, valueProvider);
+    }
+
+    public <T> void addClientLevelMutableMetric(final String name,
+                                                final String description,
+                                                final Map<String, String> additionalTags,
+                                                final RecordingLevel recordingLevel,
+                                                final Gauge<T> valueProvider) {
+        final MetricName metricName = metrics.metricName(
+            name,
+            CLIENT_LEVEL_GROUP,
+            description,
+            clientLevelTagMap(additionalTags)
+        );
         final MetricConfig metricConfig = new MetricConfig().recordLevel(recordingLevel);
         synchronized (clientLevelMetrics) {
             metrics.addMetric(metricName, metricConfig, valueProvider);
@@ -281,9 +304,12 @@ private String threadSensorPrefix(final String threadId) {
     }
 
     public Map<String, String> clientLevelTagMap() {
-        final Map<String, String> tagMap = new LinkedHashMap<>();
+        return clientLevelTagMap(Collections.emptyMap());
+    }
+
+    public Map<String, String> clientLevelTagMap(final Map<String, String> additionalTags) {
+        final Map<String, String> tagMap = new LinkedHashMap<>(additionalTags);
         tagMap.put(CLIENT_ID_TAG, clientId);
-        tagMap.put(PROCESS_ID_TAG, processId);
         return tagMap;
     }
 

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

@@ -37,6 +37,7 @@
 
 public class ClientMetricsTest {
     private static final String COMMIT_ID = "test-commit-ID";
+    private static final String PROCESS_ID = "test-process-id";
     private static final String VERSION = "test-version";
 
     private final StreamsMetricsImpl streamsMetrics = mock(StreamsMetricsImpl.class);
@@ -116,11 +117,15 @@ 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)
+
+        ClientMetrics.addClientStateTelemetryMetric(PROCESS_ID, streamsMetrics, stateProvider);
+
+        verify(streamsMetrics).addClientLevelMutableMetric(
+            eq(name),
+            eq(description),
+            eq(Collections.singletonMap("process-id", PROCESS_ID)),
+            eq(RecordingLevel.INFO),
+            eq(stateProvider)
         );
     }
 
@@ -129,11 +134,15 @@ 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)
+
+        ClientMetrics.addClientRecordingLevelMetric(PROCESS_ID, streamsMetrics, recordingLevel);
+
+        verify(streamsMetrics).addClientLevelImmutableMetric(
+            eq(name),
+            eq(description),
+            eq(Collections.singletonMap("process-id", PROCESS_ID)),
+            eq(RecordingLevel.INFO),
+            eq(recordingLevel)
         );
     }
 

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", 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", 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", time),
             time,
             "clientId",
             stateRestoreListener,
@@ -171,7 +171,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", time),
             time,
             "clientId",
             stateRestoreListener,
@@ -420,7 +420,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", 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", new MockTime());
     }
 }

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

@@ -219,7 +219,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", new MockTime());
         final InternalMockProcessorContext<Object, Object> context = new InternalMockProcessorContext<>(streamsMetrics);
         final ProcessorNode<Object, Object, Object, Object> node =
             new ProcessorNode<>(NAME, new NoOpProcessor(), Collections.emptySet());
@@ -303,7 +303,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", new MockTime());
         final InternalMockProcessorContext<Object, Object> context = new InternalMockProcessorContext<>(streamsMetrics);
         final ProcessorNode<Object, Object, Object, Object> node =
             new ProcessorNode<>("pname", new ClassCastProcessor(), Collections.emptySet());
@@ -323,7 +323,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", new MockTime()));
         when(internalProcessorContext.topic()).thenReturn(TOPIC);
         when(internalProcessorContext.partition()).thenReturn(PARTITION);
         when(internalProcessorContext.offset()).thenReturn(OFFSET);