apache · mjsax · May 25, 2024 · May 24, 2024
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java
@@ -52,7 +52,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
-import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.LongAdder;
 import java.util.function.Function;
 
 import static org.apache.kafka.common.utils.Utils.mkEntry;
@@ -95,7 +95,7 @@ public class MeteredKeyValueStore<K, V>
     private StreamsMetricsImpl streamsMetrics;
     private TaskId taskId;
 
-    protected AtomicInteger numOpenIterators = new AtomicInteger(0);
+    protected LongAdder numOpenIterators = new LongAdder();
 
     @SuppressWarnings("rawtypes")
     private final Map<Class, QueryHandler> queryHandlers =
@@ -168,7 +168,7 @@ private void registerMetrics() {
         e2eLatencySensor = StateStoreMetrics.e2ELatencySensor(taskId.toString(), metricsScope, name(), streamsMetrics);
         iteratorDurationSensor = StateStoreMetrics.iteratorDurationSensor(taskId.toString(), metricsScope, name(), streamsMetrics);
         StateStoreMetrics.addNumOpenIteratorsGauge(taskId.toString(), metricsScope, name(), streamsMetrics,
-                (config, now) -> numOpenIterators.get());
+                (config, now) -> numOpenIterators.sum());
     }
 
     protected Serde<V> prepareValueSerdeForStore(final Serde<V> valueSerde, final SerdeGetter getter) {
@@ -467,7 +467,7 @@ private MeteredKeyValueIterator(final KeyValueIterator<Bytes, byte[]> iter,
             this.iter = iter;
             this.sensor = sensor;
             this.startNs = time.nanoseconds();
-            numOpenIterators.incrementAndGet();
+            numOpenIterators.increment();
         }
 
         @Override
@@ -491,7 +491,7 @@ public void close() {
                 final long duration = time.nanoseconds() - startNs;
                 sensor.record(duration);
                 iteratorDurationSensor.record(duration);
-                numOpenIterators.decrementAndGet();
+                numOpenIterators.decrement();
             }
         }
 
@@ -515,7 +515,7 @@ private MeteredKeyValueTimestampedIterator(final KeyValueIterator<Bytes, byte[]>
             this.sensor = sensor;
             this.valueDeserializer = valueDeserializer;
             this.startNs = time.nanoseconds();
-            numOpenIterators.incrementAndGet();
+            numOpenIterators.increment();
         }
 
         @Override
@@ -539,7 +539,7 @@ public void close() {
                 final long duration = time.nanoseconds() - startNs;
                 sensor.record(duration);
                 iteratorDurationSensor.record(duration);
-                numOpenIterators.decrementAndGet();
+                numOpenIterators.decrement();
             }
         }
 

diff --git a/.../java/org/apache/kafka/streams/state/internals/MeteredMultiVersionedKeyQueryIterator.java b/.../java/org/apache/kafka/streams/state/internals/MeteredMultiVersionedKeyQueryIterator.java
@@ -16,7 +16,7 @@
  */
 package org.apache.kafka.streams.state.internals;
 
-import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.LongAdder;
 import java.util.function.Function;
 
 import org.apache.kafka.common.metrics.Sensor;
@@ -28,7 +28,7 @@ public class MeteredMultiVersionedKeyQueryIterator<V> implements VersionedRecord
 
     private final VersionedRecordIterator<byte[]> iterator;
     private final Function<VersionedRecord<byte[]>, VersionedRecord<V>> deserializeValue;
-    private final AtomicInteger numOpenIterators;
+    private final LongAdder numOpenIterators;
     private final Sensor sensor;
     private final Time time;
     private final long startNs;
@@ -37,14 +37,14 @@ public MeteredMultiVersionedKeyQueryIterator(final VersionedRecordIterator<byte[
                                                  final Sensor sensor,
                                                  final Time time,
                                                  final Function<VersionedRecord<byte[]>, VersionedRecord<V>> deserializeValue,
-                                                 final AtomicInteger numOpenIterators) {
+                                                 final LongAdder numOpenIterators) {
         this.iterator = iterator;
         this.deserializeValue = deserializeValue;
         this.numOpenIterators = numOpenIterators;
         this.sensor = sensor;
         this.time = time;
         this.startNs = time.nanoseconds();
-        numOpenIterators.incrementAndGet();
+        numOpenIterators.increment();
     }
 
 
@@ -54,7 +54,7 @@ public void close() {
             iterator.close();
         } finally {
             sensor.record(time.nanoseconds() - startNs);
-            numOpenIterators.decrementAndGet();
+            numOpenIterators.decrement();
         }
     }
 

diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredSessionStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredSessionStore.java
@@ -47,7 +47,7 @@
 
 import java.util.Map;
 import java.util.Objects;
-import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.LongAdder;
 
 import static org.apache.kafka.common.utils.Utils.mkEntry;
 import static org.apache.kafka.common.utils.Utils.mkMap;
@@ -72,7 +72,7 @@ public class MeteredSessionStore<K, V>
     private InternalProcessorContext<?, ?> context;
     private TaskId taskId;
 
-    private AtomicInteger numOpenIterators = new AtomicInteger(0);
+    private LongAdder numOpenIterators = new LongAdder();
 
     @SuppressWarnings("rawtypes")
     private final Map<Class, QueryHandler> queryHandlers =
@@ -137,7 +137,7 @@ private void registerMetrics() {
         e2eLatencySensor = StateStoreMetrics.e2ELatencySensor(taskId.toString(), metricsScope, name(), streamsMetrics);
         iteratorDurationSensor = StateStoreMetrics.iteratorDurationSensor(taskId.toString(), metricsScope, name(), streamsMetrics);
         StateStoreMetrics.addNumOpenIteratorsGauge(taskId.toString(), metricsScope, name(), streamsMetrics,
-                (config, now) -> numOpenIterators.get());
+                (config, now) -> numOpenIterators.sum());
     }
 
 

diff --git a/...c/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStore.java b/...c/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStore.java
@@ -325,7 +325,7 @@ private MeteredTimestampedKeyValueStoreIterator(final KeyValueIterator<Bytes, by
             this.valueAndTimestampDeserializer = valueAndTimestampDeserializer;
             this.startNs = time.nanoseconds();
             this.returnPlainValue = returnPlainValue;
-            numOpenIterators.incrementAndGet();
+            numOpenIterators.increment();
         }
 
         @Override
@@ -353,7 +353,7 @@ public void close() {
                 final long duration = time.nanoseconds() - startNs;
                 sensor.record(duration);
                 iteratorDurationSensor.record(duration);
-                numOpenIterators.decrementAndGet();
+                numOpenIterators.decrement();
             }
         }
 

diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredWindowStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredWindowStore.java
@@ -49,7 +49,7 @@
 
 import java.util.Map;
 import java.util.Objects;
-import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.LongAdder;
 
 import static org.apache.kafka.common.utils.Utils.mkEntry;
 import static org.apache.kafka.common.utils.Utils.mkMap;
@@ -76,7 +76,7 @@ public class MeteredWindowStore<K, V>
     private InternalProcessorContext<?, ?> context;
     private TaskId taskId;
 
-    private AtomicInteger numOpenIterators = new AtomicInteger(0);
+    private LongAdder numOpenIterators = new LongAdder();
 
     @SuppressWarnings("rawtypes")
     private final Map<Class, QueryHandler> queryHandlers =
@@ -156,7 +156,7 @@ private void registerMetrics() {
         e2eLatencySensor = StateStoreMetrics.e2ELatencySensor(taskId.toString(), metricsScope, name(), streamsMetrics);
         iteratorDurationSensor = StateStoreMetrics.iteratorDurationSensor(taskId.toString(), metricsScope, name(), streamsMetrics);
         StateStoreMetrics.addNumOpenIteratorsGauge(taskId.toString(), metricsScope, name(), streamsMetrics,
-                (config, now) -> numOpenIterators.get());
+                (config, now) -> numOpenIterators.sum());
     }
 
     @Deprecated

diff --git a/...ms/src/main/java/org/apache/kafka/streams/state/internals/MeteredWindowStoreIterator.java b/...ms/src/main/java/org/apache/kafka/streams/state/internals/MeteredWindowStoreIterator.java
@@ -22,7 +22,7 @@
 import org.apache.kafka.streams.StreamsMetrics;
 import org.apache.kafka.streams.state.WindowStoreIterator;
 
-import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.LongAdder;
 import java.util.function.Function;
 
 class MeteredWindowStoreIterator<V> implements WindowStoreIterator<V> {
@@ -34,15 +34,15 @@ class MeteredWindowStoreIterator<V> implements WindowStoreIterator<V> {
     private final Function<byte[], V> valueFrom;
     private final long startNs;
     private final Time time;
-    private final AtomicInteger numOpenIterators;
+    private final LongAdder numOpenIterators;
 
     MeteredWindowStoreIterator(final WindowStoreIterator<byte[]> iter,
                                final Sensor operationSensor,
                                final Sensor iteratorSensor,
                                final StreamsMetrics metrics,
                                final Function<byte[], V> valueFrom,
                                final Time time,
-                               final AtomicInteger numOpenIterators) {
+                               final LongAdder numOpenIterators) {
         this.iter = iter;
         this.operationSensor = operationSensor;
         this.iteratorSensor = iteratorSensor;
@@ -51,7 +51,7 @@ class MeteredWindowStoreIterator<V> implements WindowStoreIterator<V> {
         this.startNs = time.nanoseconds();
         this.time = time;
         this.numOpenIterators = numOpenIterators;
-        numOpenIterators.incrementAndGet();
+        numOpenIterators.increment();
     }
 
     @Override
@@ -73,7 +73,7 @@ public void close() {
             final long duration = time.nanoseconds() - startNs;
             operationSensor.record(duration);
             iteratorSensor.record(duration);
-            numOpenIterators.decrementAndGet();
+            numOpenIterators.decrement();
         }
     }
 

diff --git a/...c/main/java/org/apache/kafka/streams/state/internals/MeteredWindowedKeyValueIterator.java b/...c/main/java/org/apache/kafka/streams/state/internals/MeteredWindowedKeyValueIterator.java
@@ -24,7 +24,7 @@
 import org.apache.kafka.streams.kstream.Windowed;
 import org.apache.kafka.streams.state.KeyValueIterator;
 
-import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.LongAdder;
 import java.util.function.Function;
 
 class MeteredWindowedKeyValueIterator<K, V> implements KeyValueIterator<Windowed<K>, V> {
@@ -37,7 +37,7 @@ class MeteredWindowedKeyValueIterator<K, V> implements KeyValueIterator<Windowed
     private final Function<byte[], V> deserializeValue;
     private final long startNs;
     private final Time time;
-    private final AtomicInteger numOpenIterators;
+    private final LongAdder numOpenIterators;
 
     MeteredWindowedKeyValueIterator(final KeyValueIterator<Windowed<Bytes>, byte[]> iter,
                                     final Sensor operationSensor,
@@ -46,7 +46,7 @@ class MeteredWindowedKeyValueIterator<K, V> implements KeyValueIterator<Windowed
                                     final Function<byte[], K> deserializeKey,
                                     final Function<byte[], V> deserializeValue,
                                     final Time time,
-                                    final AtomicInteger numOpenIterators) {
+                                    final LongAdder numOpenIterators) {
         this.iter = iter;
         this.operationSensor = operationSensor;
         this.iteratorSensor = iteratorSensor;
@@ -56,7 +56,7 @@ class MeteredWindowedKeyValueIterator<K, V> implements KeyValueIterator<Windowed
         this.startNs = time.nanoseconds();
         this.time = time;
         this.numOpenIterators = numOpenIterators;
-        numOpenIterators.incrementAndGet();
+        numOpenIterators.increment();
     }
 
     @Override
@@ -83,7 +83,7 @@ public void close() {
             final long duration = time.nanoseconds() - startNs;
             operationSensor.record(duration);
             iteratorSensor.record(duration);
-            numOpenIterators.decrementAndGet();
+            numOpenIterators.decrement();
         }
     }
 

diff --git a/...ams/src/main/java/org/apache/kafka/streams/state/internals/metrics/StateStoreMetrics.java b/...ams/src/main/java/org/apache/kafka/streams/state/internals/metrics/StateStoreMetrics.java
@@ -438,7 +438,7 @@ public static void addNumOpenIteratorsGauge(final String taskId,
                                                 final String storeType,
                                                 final String storeName,
                                                 final StreamsMetricsImpl streamsMetrics,
-                                                final Gauge<Integer> numOpenIteratorsGauge) {
+                                                final Gauge<Long> numOpenIteratorsGauge) {
         streamsMetrics.addStoreLevelMutableMetric(
                 taskId,
                 storeType,

diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStoreTest.java
@@ -451,13 +451,13 @@ public void shouldTrackOpenIteratorsMetric() {
         final KafkaMetric openIteratorsMetric = metric("num-open-iterators");
         assertThat(openIteratorsMetric, not(nullValue()));
 
-        assertThat((Integer) openIteratorsMetric.metricValue(), equalTo(0));
+        assertThat((Long) openIteratorsMetric.metricValue(), equalTo(0L));
 
         try (final KeyValueIterator<String, String> iterator = metered.prefixScan(KEY, stringSerializer)) {
-            assertThat((Integer) openIteratorsMetric.metricValue(), equalTo(1));
+            assertThat((Long) openIteratorsMetric.metricValue(), equalTo(1L));
         }
 
-        assertThat((Integer) openIteratorsMetric.metricValue(), equalTo(0));
+        assertThat((Long) openIteratorsMetric.metricValue(), equalTo(0L));
     }
 
     @Test

diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredSessionStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredSessionStoreTest.java
@@ -615,13 +615,13 @@ public void shouldTrackOpenIteratorsMetric() {
         final KafkaMetric openIteratorsMetric = metric("num-open-iterators");
         assertThat(openIteratorsMetric, not(nullValue()));
 
-        assertThat((Integer) openIteratorsMetric.metricValue(), equalTo(0));
+        assertThat((Long) openIteratorsMetric.metricValue(), equalTo(0L));
 
         try (final KeyValueIterator<Windowed<String>, String> iterator = store.backwardFetch(KEY)) {
-            assertThat((Integer) openIteratorsMetric.metricValue(), equalTo(1));
+            assertThat((Long) openIteratorsMetric.metricValue(), equalTo(1L));
         }
 
-        assertThat((Integer) openIteratorsMetric.metricValue(), equalTo(0));
+        assertThat((Long) openIteratorsMetric.metricValue(), equalTo(0L));
     }
 
     @Test

diff --git a/...st/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreTest.java b/...st/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreTest.java
@@ -449,13 +449,13 @@ public void shouldTrackOpenIteratorsMetric() {
         final KafkaMetric openIteratorsMetric = metric("num-open-iterators");
         assertThat(openIteratorsMetric, not(nullValue()));
 
-        assertThat((Integer) openIteratorsMetric.metricValue(), equalTo(0));
+        assertThat((Long) openIteratorsMetric.metricValue(), equalTo(0L));
 
         try (final KeyValueIterator<String, ValueAndTimestamp<String>> iterator = metered.all()) {
-            assertThat((Integer) openIteratorsMetric.metricValue(), equalTo(1));
+            assertThat((Long) openIteratorsMetric.metricValue(), equalTo(1L));
         }
 
-        assertThat((Integer) openIteratorsMetric.metricValue(), equalTo(0));
+        assertThat((Long) openIteratorsMetric.metricValue(), equalTo(0L));
     }
 
     @Test

diff --git a/...test/java/org/apache/kafka/streams/state/internals/MeteredVersionedKeyValueStoreTest.java b/...test/java/org/apache/kafka/streams/state/internals/MeteredVersionedKeyValueStoreTest.java
@@ -380,15 +380,15 @@ public void shouldTrackOpenIteratorsMetric() {
         final KafkaMetric openIteratorsMetric = getMetric("num-open-iterators");
         assertThat(openIteratorsMetric, not(nullValue()));
 
-        assertThat((Integer) openIteratorsMetric.metricValue(), equalTo(0));
+        assertThat((Long) openIteratorsMetric.metricValue(), equalTo(0L));
 
         final QueryResult<VersionedRecordIterator<String>> result = store.query(query, bound, config);
 
         try (final VersionedRecordIterator<String> iterator = result.getResult()) {
-            assertThat((Integer) openIteratorsMetric.metricValue(), equalTo(1));
+            assertThat((Long) openIteratorsMetric.metricValue(), equalTo(1L));
         }
 
-        assertThat((Integer) openIteratorsMetric.metricValue(), equalTo(0));
+        assertThat((Long) openIteratorsMetric.metricValue(), equalTo(0L));
     }
 
     @Test

diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredWindowStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredWindowStoreTest.java
@@ -456,13 +456,13 @@ public void shouldTrackOpenIteratorsMetric() {
         final KafkaMetric openIteratorsMetric = metric("num-open-iterators");
         assertThat(openIteratorsMetric, not(nullValue()));
 
-        assertThat((Integer) openIteratorsMetric.metricValue(), equalTo(0));
+        assertThat((Long) openIteratorsMetric.metricValue(), equalTo(0L));
 
         try (final KeyValueIterator<Windowed<String>, String> iterator = store.all()) {
-            assertThat((Integer) openIteratorsMetric.metricValue(), equalTo(1));
+            assertThat((Long) openIteratorsMetric.metricValue(), equalTo(1L));
         }
 
-        assertThat((Integer) openIteratorsMetric.metricValue(), equalTo(0));
+        assertThat((Long) openIteratorsMetric.metricValue(), equalTo(0L));
     }
 
     @Test