This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new a578bada13 [common] Ensure ParallelExecution waits for reader cleanup
(#8855)
a578bada13 is described below
commit a578bada13856dd76171036603e3ffabe448ca0d
Author: QuakeWang <[email protected]>
AuthorDate: Thu Jul 30 10:34:49 2026 +0800
[common] Ensure ParallelExecution waits for reader cleanup (#8855)
---
.../org/apache/paimon/utils/ParallelExecution.java | 14 ++++-
.../paimon/crosspartition/IndexBootstrapTest.java | 5 --
.../apache/paimon/utils/ParallelExecutionTest.java | 67 ++++++++++++++++++++++
3 files changed, 80 insertions(+), 6 deletions(-)
diff --git
a/paimon-common/src/main/java/org/apache/paimon/utils/ParallelExecution.java
b/paimon-common/src/main/java/org/apache/paimon/utils/ParallelExecution.java
index cb8b1c1f65..2b0d2a1643 100644
--- a/paimon-common/src/main/java/org/apache/paimon/utils/ParallelExecution.java
+++ b/paimon-common/src/main/java/org/apache/paimon/utils/ParallelExecution.java
@@ -173,7 +173,19 @@ public class ParallelExecution<T, E> implements Closeable {
@Override
public void close() throws IOException {
- this.executorService.shutdownNow();
+ if (latch.getCount() == 0) {
+ this.executorService.shutdown();
+ } else {
+ this.executorService.shutdownNow();
+ }
+ try {
+ if (!this.executorService.awaitTermination(1, TimeUnit.MINUTES)) {
+ throw new IOException("Timed out while closing parallel
execution.");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IOException("Interrupted while closing parallel
execution.", e);
+ }
}
private ParallelBatch<T, E> iterator(MemorySegment page, int numRecords, E
extraMessage) {
diff --git
a/paimon-core/src/test/java/org/apache/paimon/crosspartition/IndexBootstrapTest.java
b/paimon-core/src/test/java/org/apache/paimon/crosspartition/IndexBootstrapTest.java
index 0921ddf565..0e352b63ec 100644
---
a/paimon-core/src/test/java/org/apache/paimon/crosspartition/IndexBootstrapTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/crosspartition/IndexBootstrapTest.java
@@ -105,11 +105,6 @@ public class IndexBootstrapTest extends TableTestBase {
.containsExactlyInAnyOrder(
GenericRow.of(2, 1, 3), GenericRow.of(4, 2, 5),
GenericRow.of(6, 3, 7));
result.clear();
-
- // In ParallelExecution, latch.countDown first, then close the reader,
it may not be closed
- // here, (this is good, beneficial for query speed) but
TableTestBase.after will check leak
- // streams. So sleep here to avoid unstable.
- Thread.sleep(1000);
}
private Table createTable() throws Exception {
diff --git
a/paimon-core/src/test/java/org/apache/paimon/utils/ParallelExecutionTest.java
b/paimon-core/src/test/java/org/apache/paimon/utils/ParallelExecutionTest.java
index 4e3feab578..f8348f5c5c 100644
---
a/paimon-core/src/test/java/org/apache/paimon/utils/ParallelExecutionTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/utils/ParallelExecutionTest.java
@@ -32,6 +32,13 @@ import java.util.Arrays;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.function.Supplier;
import static java.util.Collections.singletonList;
@@ -142,6 +149,66 @@ public class ParallelExecutionTest {
+ " Please increase the 'page-size' table
option.");
}
+ @Test
+ public void testCloseWaitsForReaderClose() throws Exception {
+ CountDownLatch readerCloseStarted = new CountDownLatch(1);
+ CountDownLatch readerClosed = new CountDownLatch(1);
+ Semaphore allowReaderClose = new Semaphore(0);
+ RecordReader<Integer> reader =
+ new RecordReader<Integer>() {
+ @Nullable
+ @Override
+ public RecordIterator<Integer> readBatch() {
+ return null;
+ }
+
+ @Override
+ public void close() throws IOException {
+ readerCloseStarted.countDown();
+ try {
+ allowReaderClose.acquire();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IOException("Reader close was
interrupted.", e);
+ }
+ readerClosed.countDown();
+ }
+ };
+
+ ParallelExecution<Integer, Integer> execution =
+ new ParallelExecution<>(
+ new IntSerializer(), 1024, 1, singletonList(() ->
Pair.of(reader, 1)));
+ ExecutorService closeExecutor = Executors.newSingleThreadExecutor();
+
+ try {
+ assertThat(readerCloseStarted.await(10,
TimeUnit.SECONDS)).isTrue();
+
+ CountDownLatch executionCloseStarted = new CountDownLatch(1);
+ Future<Void> closeFuture =
+ closeExecutor.submit(
+ () -> {
+ executionCloseStarted.countDown();
+ execution.close();
+ return null;
+ });
+ assertThat(executionCloseStarted.await(10,
TimeUnit.SECONDS)).isTrue();
+ assertThatThrownBy(() -> closeFuture.get(1, TimeUnit.SECONDS))
+ .isInstanceOf(TimeoutException.class);
+
+ allowReaderClose.release();
+ closeFuture.get(10, TimeUnit.SECONDS);
+ assertThat(readerClosed.await(10, TimeUnit.SECONDS)).isTrue();
+ } finally {
+ allowReaderClose.release();
+ try {
+ execution.close();
+ readerClosed.await(10, TimeUnit.SECONDS);
+ } finally {
+ closeExecutor.shutdownNow();
+ }
+ }
+ }
+
private RecordReader<Integer> create(Queue<List<Integer>> queue) {
return new RecordReader<Integer>() {
@Nullable
