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

Code before:
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.atomic.LongAdder;

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

    private final LongAdder numOpenIterators = new LongAdder();
    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);
        numOpenIterators.increment();

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

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

    public long sum() {
        return numOpenIterators.sum();
    }
}


Review comment (suggestion):
I was worried for adding the same `iterator`, which could create unexpected metrics, but it seems `StateStoreMetrics.addOldestOpenIteratorGauge` already handle this case.

Code after:
import java.util.concurrent.ConcurrentSkipListSet;

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