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

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


Review comment (suggestion):
Possible nano-optimization: the oldest start timestamp will only change if this is the first open iterator - any other open iterator will definitionally be older - so we could possibly only update the oldest timestamp inside the `if (openIterators.size() == 1)` block below?
(assuming timestamps are all coming from the same time source)

Code after:
import java.util.concurrent.atomic.AtomicLong;

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