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

Code before:
        synchronized (threadLevelSensors) {
            final Deque<String> sensors = threadLevelSensors.remove(key);
            while (sensors != null && !sensors.isEmpty()) {
                metrics.removeSensor(sensors.pop());
            }
        }
    }

    public final void removeAllThreadLevelMetrics(final String threadId) {
        synchronized (threadLevelMetrics) {
            final Deque<MetricName> names = threadLevelMetrics.remove(threadSensorPrefix(threadId));
            while (names != null && !names.isEmpty()) {
                metrics.removeMetric(names.pop());
            }
        }
    }

    public void removeStoreLevelMetric(final MetricName metricName) {
        metrics.removeMetric(metricName);
        storeLevelMetrics.get(
            storeSensorPrefix(
                metricName.tags().get("thread-id"),
                metricName.tags().get("task-id"),
                metricName.tags().get("scope-state-id")
            )
        ).remove(metricName);
    }

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

    public Map<String, String> nodeLevelTagMap(final String threadId,
                                               final String taskId,
                                               final String processorNodeName) {
        final Map<String, String> tagMap = new LinkedHashMap<>();
        tagMap.put(THREAD_ID_TAG, threadId);
        tagMap.put(TASK_ID_TAG, taskId);
        tagMap.put(PROCESSOR_NODE_ID_TAG, processorNodeName);
        return tagMap;
    }

    public Map<String, String> topicLevelTagMap(final String threadId,
                                                final String taskId,
                                                final String processorNodeName,
                                                final String topicName) {
        final Map<String, String> tagMap = new LinkedHashMap<>();
        tagMap.put(THREAD_ID_TAG, threadId);

Review comment (suggestion):
This is the actual fix. When we call `StateStoreMetrics.addOldestOpenIteratorGauge()` (inside `OpenIterators.add(...)`) to the add the metric, we not only add an entry to `metrics`, but also to `storeLevelMetrics`.

We forgot to remove the second entry...

I am not 100% sure, given the "fancy" code how we add the metric, if this removal code is 100% correct, because adding metrics is implemented in an idempotent way. -- I believe it's ok, but please don't take it for granted, and verify if this is right, or not.

Code after:
    public final void removeAllThreadLevelSensors(final String threadId) {
        final String key = threadSensorPrefix(threadId);
        synchronized (threadLevelSensors) {
            final Deque<String> sensors = threadLevelSensors.remove(key);
            while (sensors != null && !sensors.isEmpty()) {
                metrics.removeSensor(sensors.pop());
            }
        }
    }

    public final void removeAllThreadLevelMetrics(final String threadId) {
        synchronized (threadLevelMetrics) {
            final Deque<MetricName> names = threadLevelMetrics.remove(threadSensorPrefix(threadId));
            while (names != null && !names.isEmpty()) {
                metrics.removeMetric(names.pop());
            }
        }
    }

    public void removeMetric(final MetricName metricName) {
        metrics.removeMetric(metricName);
    }

    public void removeStoreLevelMetric(final MetricName metricName) {
        metrics.removeMetric(metricName);

        final List<String> metricsScopeCandidates = metricName.tags().keySet().stream()
            .filter(tag -> !tag.equals(THREAD_ID_TAG) && !tag.equals(TASK_ID_TAG))
            .collect(Collectors.toList());
        if (metricsScopeCandidates.size() != 1) {
            // should never happen
            throw new IllegalStateException("Expected exactly one metric scope tag, but found " + metricsScopeCandidates);
        }

        final Deque<MetricName> metricsForStore = storeLevelMetrics.get(
            storeSensorPrefix(
                metricName.tags().get(THREAD_ID_TAG),
                metricName.tags().get(TASK_ID_TAG),
                metricName.tags().get(metricsScopeCandidates.get(0))
            )
        );

        if (metricsForStore != null) {
            metricsForStore.remove(metricName);
        }
    }

    public Map<String, String> taskLevelTagMap(final String threadId, final String taskId) {
        final Map<String, String> tagMap = new LinkedHashMap<>();
        tagMap.put(THREAD_ID_TAG, threadId);
        tagMap.put(TASK_ID_TAG, taskId);