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

Code before:
    private final String metricsScope;
    private final String name;
    private final StreamsMetricsImpl streamsMetrics;

    private final NavigableSet<MeteredIterator> openIterators = new ConcurrentSkipListSet<>(Comparator.comparingLong(MeteredIterator::startTimestamp));

    private MetricName metricName;

    public OpenIterators(final TaskId taskId,
                         final String metricsScope,
                         final String name,
                         final StreamsMetricsImpl streamsMetrics) {
        this.taskId = taskId;
        this.metricsScope = metricsScope;
        this.name = name;
        this.streamsMetrics = streamsMetrics;
    }

    public void add(final MeteredIterator iterator) {
        openIterators.add(iterator);

        if (openIterators.size() == 1) {
            metricName = StateStoreMetrics.addOldestOpenIteratorGauge(taskId.toString(), metricsScope, name, streamsMetrics,
                (config, now) -> openIterators.first().startTimestamp()
            );
        }
    }

    public void remove(final MeteredIterator iterator) {
        if (openIterators.size() == 1) {
            streamsMetrics.removeMetric(metricName);
        }
        openIterators.remove(iterator);
    }

    public long sum() {
        return openIterators.size();
    }
}


Review comment (bug):
Thinking about this more, I wonder if there is another possible race here that would not cause an exception but could cause missing or incorrectly persisting metrics. The OpenIterators tracker itself is not synchronized (good to avoid contention) but that means that we are mutating `add()` or `remove()` and calling `size()` expecting the results in program order.
But, I don't see any guarantee this is the case - my (limited) understanding is that OpenIterators is a property of a KeyValueStore, which may have partitions shared between threads.

Imagine:

```
T1: open iterator on store S1 (partition 0)
T2: open iterator on store S1 (partition 1)
T1: openIterators.add(it1)
T2: openIterators.add(it2)
T1: if (openIterators.size() == 1) {  (2 == 1? nope)
T2: if (openIterators.size() == 1) {  (2 == 1? nope)
End: 2 iterators open, no metric registered
```

Same race is possible with remove.

Code after:
public class OpenIterators {
    private final TaskId taskId;
    private final String metricsScope;
    private final String name;
    private final StreamsMetricsImpl streamsMetrics;

    private final NavigableSet<MeteredIterator> openIterators = new ConcurrentSkipListSet<>(Comparator.comparingLong(MeteredIterator::startTimestamp));
    private final AtomicLong oldestStartTimestamp = new AtomicLong();

    private MetricName metricName;

    public OpenIterators(final TaskId taskId,
                         final String metricsScope,
                         final String name,
                         final StreamsMetricsImpl streamsMetrics) {
        this.taskId = taskId;
        this.metricsScope = metricsScope;
        this.name = name;
        this.streamsMetrics = streamsMetrics;
    }

    public void add(final MeteredIterator iterator) {
        openIterators.add(iterator);
        updateOldestStartTimestamp();

        if (openIterators.size() == 1) {
            metricName = StateStoreMetrics.addOldestOpenIteratorGauge(taskId.toString(), metricsScope, name, streamsMetrics,
                (config, now) -> oldestStartTimestamp.get()
            );
        }
    }

    public void remove(final MeteredIterator iterator) {
        if (openIterators.size() == 1) {
            streamsMetrics.removeMetric(metricName);
        }
        openIterators.remove(iterator);
        updateOldestStartTimestamp();
    }

    public long sum() {
        return openIterators.size();
    }

    private void updateOldestStartTimestamp() {
        final Iterator<MeteredIterator> openIteratorsIterator = openIterators.iterator();
        if (openIteratorsIterator.hasNext()) {
            oldestStartTimestamp.set(openIteratorsIterator.next().startTimestamp());
        }
    }
}