(paimon) branch master updated: [api] Fix race condition causing queued tasks to execute after SequentialBatchIterator close (#9452)

[lzljs3620320]

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 72caec1dff [api] Fix race condition causing queued tasks to execute 
after SequentialBatchIterator close (#9452)
72caec1dff is described below

commit 72caec1dffe66e02fe660545cc7cde535cbbcb11
Author: Juntao Zhang <juntzh...@foxmail.com>
AuthorDate: Sat Aug 29 20:55:26 2026 +0800

    [api] Fix race condition causing queued tasks to execute after 
SequentialBatchIterator close (#9452)
---
 .../org/apache/paimon/utils/ThreadPoolUtils.java   | 24 +++++++++++++++++-----
 1 file changed, 19 insertions(+), 5 deletions(-)

diff --git 
a/paimon-api/src/main/java/org/apache/paimon/utils/ThreadPoolUtils.java 
b/paimon-api/src/main/java/org/apache/paimon/utils/ThreadPoolUtils.java
index b5c28a19a0..b7151afbbd 100644
--- a/paimon-api/src/main/java/org/apache/paimon/utils/ThreadPoolUtils.java
+++ b/paimon-api/src/main/java/org/apache/paimon/utils/ThreadPoolUtils.java
@@ -229,7 +229,7 @@ public class ThreadPoolUtils {
 
         private Iterator<T> activeResults = 
Collections.<T>emptyList().iterator();
         private T next;
-        private boolean closed;
+        private volatile boolean closed;
 
         private SequentialBatchIterator(
                 ExecutorService executor,
@@ -243,7 +243,7 @@ public class ThreadPoolUtils {
 
         @Override
         public boolean hasNext() {
-            if (!closed) {
+            if (!isClosed()) {
                 advanceIfNeeded();
             }
             return next != null;
@@ -259,6 +259,10 @@ public class ThreadPoolUtils {
             return result;
         }
 
+        private boolean isClosed() {
+            return closed;
+        }
+
         private void advanceIfNeeded() {
             while (next == null) {
                 if (activeResults.hasNext()) {
@@ -286,7 +290,7 @@ public class ThreadPoolUtils {
         private void submitBatch(List<U> batch) {
             ClassLoader classLoader = 
Thread.currentThread().getContextClassLoader();
             for (U input : batch) {
-                BatchTask<T, U> task = new BatchTask<>(processor, input, 
classLoader);
+                BatchTask<T, U> task = new BatchTask<>(this, processor, input, 
classLoader);
                 executor.execute(task);
                 activeTasks.add(task);
             }
@@ -343,6 +347,7 @@ public class ThreadPoolUtils {
         private static final int CANCELLED = 2;
         private static final int FINISHED = 3;
 
+        private final SequentialBatchIterator<T, U> iterator;
         private final Function<U, List<T>> processor;
         private final U input;
         private final ClassLoader classLoader;
@@ -354,7 +359,12 @@ public class ThreadPoolUtils {
         private Throwable failure;
         private volatile boolean failureReported;
 
-        private BatchTask(Function<U, List<T>> processor, U input, ClassLoader 
classLoader) {
+        private BatchTask(
+                SequentialBatchIterator<T, U> iterator,
+                Function<U, List<T>> processor,
+                U input,
+                ClassLoader classLoader) {
+            this.iterator = iterator;
             this.processor = processor;
             this.input = input;
             this.classLoader = classLoader;
@@ -363,7 +373,7 @@ public class ThreadPoolUtils {
         @Override
         public void run() {
             synchronized (this) {
-                if (state == CANCELLED) {
+                if (isCanceled()) {
                     state = FINISHED;
                     completion.countDown();
                     return;
@@ -386,6 +396,10 @@ public class ThreadPoolUtils {
             }
         }
 
+        private boolean isCanceled() {
+            return state == CANCELLED || iterator.isClosed();
+        }
+
         private synchronized void cancel() {
             if (state == CREATED) {
                 state = CANCELLED;