fix: 11746: Backport the fix for #11304 to release 0.47 (#11747)

Signed-off-by: Artem Ananev <[email protected]>
hashgraph · Feb 26, 2024 · 62d8476 · 62d8476
1 parent 08372c4
commit 62d8476
Show file tree

Hide file tree

Showing 2 changed files with 29 additions and 8 deletions.
diff --git a/...p/src/main/java/com/swirlds/virtualmap/internal/reconnect/ConcurrentBlockingIterator.java b/...p/src/main/java/com/swirlds/virtualmap/internal/reconnect/ConcurrentBlockingIterator.java
@@ -109,17 +109,15 @@ public boolean hasNext() {
         // if closed and buffer.poll != null || !closed
         final long waitMillis = maxWaitTimeUnit.toMillis(maxWaitTime);
         final long timeOutWhenMillisAre = System.currentTimeMillis() + waitMillis;
-        while ((next = buffer.poll()) == null) {
-            if (closed.get()) {
-                return false;
-            } else {
-                if (System.currentTimeMillis() > timeOutWhenMillisAre) {
-                    throw new RuntimeException(new TimeoutException("Timed out trying to read from buffer"));
-                }
+        boolean isOpen = !closed.get();
+        while (((next = buffer.poll()) == null) && isOpen) {
+            if (System.currentTimeMillis() > timeOutWhenMillisAre) {
+                throw new RuntimeException(new TimeoutException("Timed out trying to read from buffer"));
             }
+            isOpen = !closed.get();
         }
 
-        return true;
+        return next != null;
     }
 
     /**

diff --git a/...c/test/java/com/swirlds/virtualmap/internal/reconnect/ConcurrentBlockingIteratorTest.java b/...c/test/java/com/swirlds/virtualmap/internal/reconnect/ConcurrentBlockingIteratorTest.java
@@ -31,6 +31,7 @@
 import java.util.concurrent.Future;
 import java.util.concurrent.atomic.AtomicBoolean;
 import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.RepeatedTest;
 import org.junit.jupiter.api.Test;
 
 public class ConcurrentBlockingIteratorTest {
@@ -217,6 +218,28 @@ void multiThreadIteration() throws ExecutionException, InterruptedException {
         }
     }
 
+    @RepeatedTest(1000)
+    @DisplayName("yield all elements even after closed")
+    void supplyBeforeClose() throws InterruptedException {
+        final ConcurrentBlockingIterator<Integer> iterator = new ConcurrentBlockingIterator<>(4, 10, SECONDS);
+        final Thread supplier = new Thread(() -> {
+            try {
+                iterator.supply(1);
+                iterator.supply(2);
+                iterator.close();
+            } catch (final InterruptedException e) {
+                fail("Supplier interrupted");
+            }
+        });
+        supplier.start();
+        assertTrue(iterator.hasNext(), "Iterator must have more than zero elements");
+        assertEquals(1, iterator.next());
+        assertTrue(iterator.hasNext(), "Iterator must have more than one element");
+        assertEquals(2, iterator.next());
+        assertFalse(iterator.hasNext(), "Iterator must not have more than two elements");
+        supplier.join();
+    }
+
     private <T> Future<Class<T>> thrownByCall(Runnable lambda) {
         return threadPool.submit(() -> {
             try {