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) -> {
                    final Iterator<MeteredIterator> openIteratorsIterator = openIterators.iterator();
                    return openIteratorsIterator.hasNext() ? openIteratorsIterator.next().startTimestamp() : null;
                }
            );
        }
    }

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

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


Review comment (performance):
I think we we ever want to return `null` -- that why we added dynamic register/deregister behavior.

I believe, we would need to somehow cache the "last result" before we remove the last iterator.

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