(hop) branch main updated: issue #7742: flush last batch before marking BlockingBatchingRowSet done (#7745)

[mc...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

mattcasters pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hop.git


The following commit(s) were added to refs/heads/main by this push:
     new dcee613882 issue #7742: flush last batch before marking BlockingBatchingRowSet done (#7745)
dcee613882 is described below

commit dcee61388286a3a38c5ad6607bce11940a52110e
Author: Matt Casters <ma...@gmail.com>
AuthorDate: Sun Aug 2 14:04:57 2026 +0200

    issue #7742: flush last batch before marking BlockingBatchingRowSet done (#7745)
    
    Enqueue any remaining partial batch before setDone() sets the done flag,
    so concurrent consumers cannot observe isDone() and exit before the final
    rows are available.
---
 .../apache/hop/core/BlockingBatchingRowSet.java    |   4 +-
 .../hop/core/BlockingBatchingRowSetTest.java       | 210 +++++++++++++++++++++
 2 files changed, 213 insertions(+), 1 deletion(-)

diff --git a/core/src/main/java/org/apache/hop/core/BlockingBatchingRowSet.java b/core/src/main/java/org/apache/hop/core/BlockingBatchingRowSet.java
index 88fca37851..fda80f622c 100644
--- a/core/src/main/java/org/apache/hop/core/BlockingBatchingRowSet.java
+++ b/core/src/main/java/org/apache/hop/core/BlockingBatchingRowSet.java
@@ -162,7 +162,8 @@ public class BlockingBatchingRowSet extends BaseRowSet implements Comparable<IRo
 
   @Override
   public void setDone() {
-    super.setDone();
+    // Enqueue the last partial batch before marking done so concurrent consumers that observe
+    // isDone()==true cannot exit before the final rows are available (issue #7742).
     if (putIndex > 0 && putIndex < size && inputBuffer != null) {
       inputBuffer[putIndex] = null; // signal the end of buffer
       for (int i = putIndex + 1; i < size; i++) {
@@ -171,6 +172,7 @@ public class BlockingBatchingRowSet extends BaseRowSet implements Comparable<IRo
       getArray.offer(inputBuffer);
     }
     putArray.clear();
+    super.setDone();
   }
 
   @Override
diff --git a/core/src/test/java/org/apache/hop/core/BlockingBatchingRowSetTest.java b/core/src/test/java/org/apache/hop/core/BlockingBatchingRowSetTest.java
index 09f245aaf2..e632c544ae 100644
--- a/core/src/test/java/org/apache/hop/core/BlockingBatchingRowSetTest.java
+++ b/core/src/test/java/org/apache/hop/core/BlockingBatchingRowSetTest.java
@@ -24,8 +24,16 @@ import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
+import java.lang.reflect.Field;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
 import org.apache.hop.core.row.IRowMeta;
 import org.apache.hop.core.row.IValueMeta;
 import org.apache.hop.core.row.RowMeta;
@@ -118,4 +126,206 @@ class BlockingBatchingRowSetTest {
     assertNotNull(r);
     assertArrayEquals(rows.get(2), r);
   }
+
+  /**
+   * Issue #7742: the last partial batch must be enqueued before the done flag is set, otherwise a
+   * concurrent consumer can observe isDone() and exit before the last rows are available.
+   */
+  @Test
+  void testSetDoneEnqueuesPartialBatchBeforeDoneFlag() throws Exception {
+    BlockingBatchingRowSet set = new BlockingBatchingRowSet(10);
+    IRowMeta rm = createRowMetaInterface();
+
+    // 3 rows with buffer size 5 = partial batch (not flushed by putRow)
+    for (int i = 0; i < 3; i++) {
+      assertTrue(set.putRow(rm, new Object[] {(long) i}));
+    }
+    assertEquals(0, set.size());
+
+    AtomicBoolean offerSeen = new AtomicBoolean(false);
+    AtomicBoolean doneAtOffer = new AtomicBoolean(true); // fail closed if offer never checked
+    instrumentGetArrayOffers(
+        set,
+        () -> {
+          offerSeen.set(true);
+          doneAtOffer.set(set.isDone());
+        });
+
+    set.setDone();
+
+    assertTrue(offerSeen.get(), "partial batch must be offered in setDone()");
+    assertFalse(
+        doneAtOffer.get(),
+        "done flag must not be set before the last batch is offered (issue #7742)");
+    assertTrue(set.isDone());
+  }
+
+  /** When there is nothing left to flush, setDone() should only mark the rowset done. */
+  @Test
+  void testSetDoneWithNoPartialBatchOnlyMarksDone() throws Exception {
+    BlockingBatchingRowSet set = new BlockingBatchingRowSet(10);
+    AtomicBoolean offerSeen = new AtomicBoolean(false);
+    instrumentGetArrayOffers(set, () -> offerSeen.set(true));
+
+    set.setDone();
+
+    assertFalse(offerSeen.get(), "no batch should be offered when nothing was put");
+    assertTrue(set.isDone());
+  }
+
+  /**
+   * When the last put already filled and published a full batch, setDone() must not re-offer; only
+   * the done flag is set.
+   */
+  @Test
+  void testSetDoneAfterFullBatchOnlyMarksDone() throws Exception {
+    BlockingBatchingRowSet set = new BlockingBatchingRowSet(10);
+    IRowMeta rm = createRowMetaInterface();
+
+    // Exactly one full batch (size = maxSize / BATCHSIZE = 5)
+    for (int i = 0; i < 5; i++) {
+      assertTrue(set.putRow(rm, new Object[] {(long) i}));
+    }
+    assertEquals(5, set.size());
+
+    AtomicBoolean offerSeen = new AtomicBoolean(false);
+    instrumentGetArrayOffers(set, () -> offerSeen.set(true));
+
+    set.setDone();
+
+    assertFalse(offerSeen.get(), "full batch was already published; setDone must not re-offer");
+    assertTrue(set.isDone());
+  }
+
+  /**
+   * After setDone() with a partial batch, all non-null rows must be readable (sequential drain).
+   */
+  @Test
+  void testPartialBatchDrainableAfterSetDone() {
+    BlockingBatchingRowSet set = new BlockingBatchingRowSet(10);
+    IRowMeta rm = createRowMetaInterface();
+    final int rowCount = 3;
+
+    for (int i = 0; i < rowCount; i++) {
+      assertTrue(set.putRow(rm, new Object[] {(long) i}));
+    }
+
+    set.setDone();
+    assertTrue(set.isDone());
+
+    List<Object[]> received = drainNonNullRows(set);
+    assertEquals(rowCount, received.size());
+    for (int i = 0; i < rowCount; i++) {
+      assertArrayEquals(new Object[] {(long) i}, received.get(i));
+    }
+  }
+
+  /**
+   * Concurrent producer/consumer using the same done-check pattern as BaseTransform.getRow().
+   * Partial batches must never be lost across repeated runs (issue #7742).
+   */
+  @Test
+  void testNoDataLossOnConcurrentSetDone() throws Exception {
+    final int iterations = 200;
+    final int rowCount = 3;
+    IRowMeta rm = createRowMetaInterface();
+
+    for (int iter = 0; iter < iterations; iter++) {
+      BlockingBatchingRowSet set = new BlockingBatchingRowSet(10);
+      List<Object[]> received = Collections.synchronizedList(new ArrayList<>());
+      CountDownLatch consumerStarted = new CountDownLatch(1);
+      AtomicReference<Throwable> consumerError = new AtomicReference<>();
+
+      Thread consumer =
+          new Thread(
+              () -> {
+                try {
+                  consumerStarted.countDown();
+                  // Mirrors BaseTransform handling when getRowWait returns null
+                  while (true) {
+                    Object[] row = set.getRowWait(20, TimeUnit.MILLISECONDS);
+                    if (row != null) {
+                      received.add(row);
+                    } else if (set.isDone()) {
+                      row = set.getRowWait(1, TimeUnit.MILLISECONDS);
+                      if (row == null) {
+                        break;
+                      }
+                      received.add(row);
+                    }
+                  }
+                } catch (Throwable t) {
+                  consumerError.set(t);
+                }
+              },
+              "batching-rowset-consumer-" + iter);
+
+      consumer.start();
+      assertTrue(consumerStarted.await(5, TimeUnit.SECONDS));
+
+      for (int i = 0; i < rowCount; i++) {
+        assertTrue(set.putRow(rm, new Object[] {(long) i}));
+      }
+      set.setDone();
+
+      consumer.join(10_000);
+      assertFalse(consumer.isAlive(), "consumer did not finish at iteration " + iter);
+      assertNull(consumerError.get(), () -> "consumer failed: " + consumerError.get());
+      assertEquals(
+          rowCount,
+          received.size(),
+          "data loss at iteration " + iter + ", got " + received.size() + " rows");
+      for (int i = 0; i < rowCount; i++) {
+        assertArrayEquals(new Object[] {(long) i}, received.get(i));
+      }
+    }
+  }
+
+  /**
+   * Replace getArray with a queue that invokes {@code onOffer} for every subsequent offer. Existing
+   * batches are preserved without firing the callback.
+   */
+  private static void instrumentGetArrayOffers(BlockingBatchingRowSet set, Runnable onOffer)
+      throws Exception {
+    Field getArrayField = BlockingBatchingRowSet.class.getDeclaredField("getArray");
+    getArrayField.setAccessible(true);
+    @SuppressWarnings("unchecked")
+    BlockingQueue<Object[][]> originalGetArray = (BlockingQueue<Object[][]>) getArrayField.get(set);
+
+    AtomicBoolean logOffers = new AtomicBoolean(false);
+    BlockingQueue<Object[][]> loggingGetArray =
+        new ArrayBlockingQueue<>(Math.max(2, originalGetArray.size() + 2), true) {
+          @Override
+          public boolean offer(Object[][] e) {
+            if (logOffers.get()) {
+              onOffer.run();
+            }
+            return super.offer(e);
+          }
+
+          @Override
+          public boolean offer(Object[][] e, long timeout, TimeUnit unit)
+              throws InterruptedException {
+            if (logOffers.get()) {
+              onOffer.run();
+            }
+            return super.offer(e, timeout, unit);
+          }
+        };
+    // Preserve any batches already published (drainTo uses offer; logging still off)
+    originalGetArray.drainTo(loggingGetArray);
+    getArrayField.set(set, loggingGetArray);
+    logOffers.set(true);
+  }
+
+  private static List<Object[]> drainNonNullRows(BlockingBatchingRowSet set) {
+    List<Object[]> received = new ArrayList<>();
+    Object[] row;
+    // Partial batches pad with null sentinels; stop after the first null once we have started
+    // seeing data, but keep polling while empty before setDone flush visibility in sequential use.
+    while ((row = set.getRowWait(100, TimeUnit.MILLISECONDS)) != null) {
+      received.add(row);
+    }
+    return received;
+  }
 }