Re: [PR] KAFKA-14460: Skip removed entries from in-memory KeyValueIterator [kafka]

["ableegoldman (via GitHub)" <gi...@apache.org>]

ableegoldman commented on code in PR #16505:
URL: https://github.com/apache/kafka/pull/16505#discussion_r1680397701


##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStore.java:
##########
@@ -265,23 +264,45 @@ private InMemoryKeyValueIterator(final Set<Bytes> keySet, final boolean forward)
 
         @Override
         public boolean hasNext() {
-            return iter.hasNext();
+            if (!open) {
+                throw new InvalidStateStoreException(String.format("Store %s has closed", name));

Review Comment:
   This is going to hit the `open` field of the iterator, not the store itself which I think is what you were going for. There's not really an reason to track whether the iterator is open or not so I'd say you can just remove the additional `private volatile Boolean open` field that you added, and then we'll be good. Make sense?



##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStore.java:
##########
@@ -265,23 +264,45 @@ private InMemoryKeyValueIterator(final Set<Bytes> keySet, final boolean forward)
 
         @Override
         public boolean hasNext() {
-            return iter.hasNext();
+            if (!open) {
+                throw new InvalidStateStoreException(String.format("Store %s has closed", name));
+            }
+            if (currentKey != null) {
+                if (map.containsKey(currentKey)) {
+                    return true;
+                } else {
+                    currentKey = null;
+                    return hasNext();
+                }
+            }
+            if (!iter.hasNext()) {
+                return false;
+            }
+            currentKey = iter.next();
+            return hasNext();
         }
 
         @Override
         public KeyValue<Bytes, byte[]> next() {
-            final Bytes key = iter.next();
-            return new KeyValue<>(key, map.get(key));
+            if (!hasNext()) {
+                throw new NoSuchElementException();
+            }
+            final KeyValue<Bytes, byte[]> ret = new KeyValue<>(currentKey, map.get(currentKey));
+            currentKey = null;
+            return ret;
         }
 
         @Override
         public void close() {
-            // do nothing
+            open = false;

Review Comment:
   We should remove this as well when we remove the local `open` field in the iterator class so we don't accidentally close the store itself



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org