[jira] [Commented] (KAFKA-18689) NoSuchElementException in state store iterator metrics

["Matthias J. Sax (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/KAFKA-18689?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17922880#comment-17922880 ] 

Matthias J. Sax commented on KAFKA-18689:
-----------------------------------------

Thanks for filing the ticket. Yes, it's a duplicate of https://issues.apache.org/jira/browse/KAFKA-17954 – Let me take a look at your PR. If we can indeed improve the fix, happy to merge it.

> NoSuchElementException in state store iterator metrics
> ------------------------------------------------------
>
>                 Key: KAFKA-18689
>                 URL: https://issues.apache.org/jira/browse/KAFKA-18689
>             Project: Kafka
>          Issue Type: Bug
>          Components: metrics, streams
>    Affects Versions: 3.9.0
>            Reporter: Steven Schlansker
>            Priority: Minor
>
> We are standing up a new Kafka Streams app that collects metrics and extensively uses state store iterators.
>  
> Occasionally, we see the following exception in our logs:
> {code:java}
> java.util.NoSuchElementException: null
> 	at java.base/java.util.concurrent.ConcurrentSkipListMap.firstKey(ConcurrentSkipListMap.java:1863)
> 	at java.base/java.util.concurrent.ConcurrentSkipListSet.first(ConcurrentSkipListSet.java:398)
> 	at org.apache.kafka.streams.state.internals.MeteredKeyValueStore.lambda$registerMetrics$5(MeteredKeyValueStore.java:176)
> 	at org.apache.kafka.common.metrics.KafkaMetric.metricValue(KafkaMetric.java:81)
> 	at com.paywholesail.service.search.core.streams.KafkaStreamsMetricsManager$1.getValue(KafkaStreamsMetricsManager.java:53)
> 	at io.dropwizard.metrics5.graphite.GraphiteReporter.reportGauge(GraphiteReporter.java:480)
> 	at io.dropwizard.metrics5.graphite.GraphiteReporter.report(GraphiteReporter.java:378) {code}
> Looking at the implementation of state store metrics (MeteredKeyValueStore.registerMetrics),
> {code:java}
> StateStoreMetrics.addOldestOpenIteratorGauge(taskId.toString(), metricsScope, name(), streamsMetrics,
>         (config, now) -> openIterators.isEmpty() ? null : openIterators.first().startTimestamp()
> );{code}
> OpenIterators is a ConcurrentSkipListSet and is not protected by any lock while the metric is read. Therefore, checking isEmpty and calling first has a natural check-then-use race condition, and throws an exception if the last open iterator is removed concurrently.
>  
> It might be safer to always take an iterator, like:
> {code:java}
> var openIterIter = openIterators.iterator();
> return openIterIter.hasNext() ? openIterIter.next().startTimestamp() : null; {code}
>  



--
This message was sent by Atlassian Jira
(v8.20.10#820010)