Repository: apache/kafka
File: streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/StreamsMetricsImpl.java (Java)

Code before:
        }
    }

    public final Sensor threadLevelSensor(final String threadId,
                                          final String sensorSuffix,
                                          final RecordingLevel recordingLevel,
                                          final Sensor... parents) {
        final String sensorPrefix = threadSensorPrefix(threadId);
        synchronized (threadLevelSensors) {
            return getSensors(threadLevelSensors, sensorSuffix, sensorPrefix, recordingLevel, parents);
        }
    }

    private String threadSensorPrefix(final String threadId) {
        return SENSOR_INTERNAL_LABEL + SENSOR_PREFIX_DELIMITER + threadId;
    }

    public Map<String, String> clientLevelTagMap() {
        final Map<String, String> tagMap = new LinkedHashMap<>();
        tagMap.put(CLIENT_ID_TAG, clientId);
        tagMap.put(PROCESS_ID_TAG, processId);
        return tagMap;
    }

    public Map<String, String> threadLevelTagMap(final String threadId) {
        return threadLevelTagMap(threadId, Collections.emptyMap());
    }

    public Map<String, String> threadLevelTagMap(final String threadId, final Map<String, String> additionalTags) {
        final Map<String, String> tagMap = new LinkedHashMap<>(additionalTags);
        tagMap.put(THREAD_ID_TAG, threadId);
        return tagMap;
    }

    public final void removeAllClientLevelSensorsAndMetrics() {
        removeAllClientLevelSensors();
        removeAllClientLevelMetrics();
    }

    private void removeAllClientLevelMetrics() {
        synchronized (clientLevelMetrics) {
            while (!clientLevelMetrics.isEmpty()) {
                metrics.removeMetric(clientLevelMetrics.pop());
            }
        }
    }

    private void removeAllClientLevelSensors() {
        synchronized (clientLevelSensors) {
            while (!clientLevelSensors.isEmpty()) {
                metrics.removeSensor(clientLevelSensors.pop());

Review comment (suggestion):
Pardon me. I see this new tag applied al all metrics, yet the KIP only specifies the `client-state` metric. Am I misunderstanding the scope? thanks!

Code after:
                clientLevelSensors.push(fullSensorName);
                return metrics.sensor(fullSensorName, recordingLevel, parents);
            }
            return sensor;
        }
    }

    public final Sensor threadLevelSensor(final String threadId,
                                          final String sensorSuffix,
                                          final RecordingLevel recordingLevel,
                                          final Sensor... parents) {
        final String sensorPrefix = threadSensorPrefix(threadId);
        synchronized (threadLevelSensors) {
            return getSensors(threadLevelSensors, sensorSuffix, sensorPrefix, recordingLevel, parents);
        }
    }

    private String threadSensorPrefix(final String threadId) {
        return SENSOR_INTERNAL_LABEL + SENSOR_PREFIX_DELIMITER + threadId;
    }

    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;
    }

    public Map<String, String> threadLevelTagMap(final String threadId) {
        return threadLevelTagMap(threadId, Collections.emptyMap());
    }

    public Map<String, String> threadLevelTagMap(final String threadId, final Map<String, String> additionalTags) {
        final Map<String, String> tagMap = new LinkedHashMap<>(additionalTags);
        tagMap.put(THREAD_ID_TAG, threadId);
        return tagMap;
    }

    public final void removeAllClientLevelSensorsAndMetrics() {
        removeAllClientLevelSensors();
        removeAllClientLevelMetrics();
    }

    private void removeAllClientLevelMetrics() {
        synchronized (clientLevelMetrics) {
            while (!clientLevelMetrics.isEmpty()) {
                metrics.removeMetric(clientLevelMetrics.pop());
            }
        }
    }