MINOR: Use thread name and task for sensor name

[bbejeck]

Changes to keep the operation name as is and make the sensor name unique.

Committer Checklist (excluded from commit message)

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

[bbejeck]

@guozhangwang @mjsax @vvcephei

[mjsax]

This is a follow up to #5009 that incorrectly changes not only sensor names, but also metric names that should not change.

[mjsax]

LGTM.

[vvcephei]

What do you think about this instead (so we can avoid mutable taskName in this multithreaded class):

diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/StreamsMetricsImpl.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/StreamsMetricsImpl.java
index bc2e15012..6a3b22cc2 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/StreamsMetricsImpl.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/StreamsMetricsImpl.java
@@ -253,12 +253,30 @@ public class StreamsMetricsImpl implements StreamsMetrics {
                                                 final String operationName,
                                                 final Sensor.RecordingLevel recordingLevel,
                                                 final String... tags) {
-        return addLatencyAndThroughputSensor(
-            scopeName,
-            entityName,
-            threadName + "." + taskName + "." + operationName,
+
+        final Map<String, String> tagMap = constructTags(scopeName, entityName, tags);
+        final Map<String, String> allTagMap = constructTags(scopeName, "all", tags);
+
+        // first add the global operation metrics if not yet, with the global tags only
+        final Sensor parent = metrics.sensor(
+            sensorName(threadName + "." + taskName + "." + operationName, null),
+            recordingLevel
+        );
+        addLatencyMetrics(scopeName, parent, operationName, allTagMap);
+        addThroughputMetrics(scopeName, parent, operationName, allTagMap);
+
+        // add the operation metrics with additional tags
+        final Sensor sensor = metrics.sensor(
+            sensorName(threadName + "." + taskName + "." + operationName, entityName),
             recordingLevel,
-            tags);
+            parent
+        );
+        addLatencyMetrics(scopeName, sensor, operationName, tagMap);
+        addThroughputMetrics(scopeName, sensor, operationName, tagMap);
+
+        parentSensors.put(sensor, parent);
+
+        return sensor;
     }
 
     /**
@@ -292,13 +310,27 @@ public class StreamsMetricsImpl implements StreamsMetrics {
                                       final String operationName,
                                       final Sensor.RecordingLevel recordingLevel,
                                       final String... tags) {
-        return addThroughputSensor(
-            scopeName,
-            entityName,
-            threadName + "." + taskName + "." + operationName,
+        final Map<String, String> tagMap = constructTags(scopeName, entityName, tags);
+        final Map<String, String> allTagMap = constructTags(scopeName, "all", tags);
+
+        // first add the global operation metrics if not yet, with the global tags only
+        final Sensor parent = metrics.sensor(
+            sensorName(threadName + "." + taskName + "." + operationName, null),
+            recordingLevel
+        );
+        addThroughputMetrics(scopeName, parent, operationName, allTagMap);
+
+        // add the operation metrics with additional tags
+        final Sensor sensor = metrics.sensor(
+            sensorName(threadName + "." + taskName + "." + operationName, entityName),
             recordingLevel,
-            tags
+            parent
         );
+        addThroughputMetrics(scopeName, sensor, operationName, tagMap);
+
+        parentSensors.put(sensor, parent);
+
+        return sensor;
     }

[vvcephei]

L116 and 118 also have changes that should be reverted.

[bbejeck]

hmm, I'm not sure about that. Those lines are getting the sensor by name which still uses the combination of thread.name() + taskID + operationName or am I missing something?

[vvcephei]

oh! yeah, you're right. Sorry!

[vvcephei]

Thanks for the patch, @bbejeck.

I'm concerned about the unprotected mutable state this introduces. If multithreaded access to the methods in question occurs, then the task name may be overwritten by the second caller before the first caller creates the sensor name, resulting in incorrect execution.

Also, there is one place where we need to revert a couple more lines from #5009 to return the tests to their original state.

[bbejeck]

updated this.

@mjsax @vvcephei have another look and let me know your thoughts on this change.

[vvcephei]

I think you meant addLatencyAndThroughputSensor

[bbejeck]

doh! yes, good catch

[vvcephei]

There's one typo, otherwise LGTM! Thanks again.

[guozhangwang]

LGTM! Thanks @bbejeck