(paimon) branch master updated: [core][test] Fix flaky thread pool submission assertion

[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 592ea42941 [core][test] Fix flaky thread pool submission assertion
592ea42941 is described below

commit 592ea42941def0cbb83623b04e3231b133ae4659
Author: JingsongLi <jingsongl...@gmail.com>
AuthorDate: Wed Aug 26 22:43:07 2026 +0800

    [core][test] Fix flaky thread pool submission assertion
---
 .../apache/paimon/utils/ThreadPoolUtilsTest.java   | 33 +++++++++++++++++++---
 1 file changed, 29 insertions(+), 4 deletions(-)

diff --git 
a/paimon-api/src/test/java/org/apache/paimon/utils/ThreadPoolUtilsTest.java 
b/paimon-api/src/test/java/org/apache/paimon/utils/ThreadPoolUtilsTest.java
index 65a3331fc7..4bb556343f 100644
--- a/paimon-api/src/test/java/org/apache/paimon/utils/ThreadPoolUtilsTest.java
+++ b/paimon-api/src/test/java/org/apache/paimon/utils/ThreadPoolUtilsTest.java
@@ -30,6 +30,7 @@ import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
+import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -44,7 +45,7 @@ public class ThreadPoolUtilsTest {
 
     @Test
     public void testCloseableBatchReturnsInOrderAndBoundsSubmission() throws 
Exception {
-        ThreadPoolExecutor workers = (ThreadPoolExecutor) 
Executors.newFixedThreadPool(2);
+        CountingThreadPoolExecutor workers = new CountingThreadPoolExecutor(2);
         ExecutorService consumer = Executors.newSingleThreadExecutor();
         CountDownLatch firstStarted = new CountDownLatch(1);
         CountDownLatch secondFinished = new CountDownLatch(1);
@@ -74,7 +75,7 @@ public class ThreadPoolUtilsTest {
 
             assertThat(firstStarted.await(3, TimeUnit.SECONDS)).isTrue();
             assertThat(secondFinished.await(3, TimeUnit.SECONDS)).isTrue();
-            assertThat(workers.getTaskCount()).isEqualTo(2);
+            assertThat(workers.getSubmittedTaskCount()).isEqualTo(2);
             assertThat(firstResult.isDone()).isFalse();
 
             releaseFirst.countDown();
@@ -82,10 +83,10 @@ public class ThreadPoolUtilsTest {
             results.add(firstResult.get(3, TimeUnit.SECONDS));
             assertThat(iterator.hasNext()).isTrue();
             results.add(iterator.next());
-            assertThat(workers.getTaskCount()).isEqualTo(2);
+            assertThat(workers.getSubmittedTaskCount()).isEqualTo(2);
 
             assertThat(iterator.hasNext()).isTrue();
-            assertThat(workers.getTaskCount()).isEqualTo(4);
+            assertThat(workers.getSubmittedTaskCount()).isEqualTo(4);
             results.add(iterator.next());
             assertThat(iterator.hasNext()).isTrue();
             results.add(iterator.next());
@@ -235,4 +236,28 @@ public class ThreadPoolUtilsTest {
             }
         }
     }
+
+    private static class CountingThreadPoolExecutor extends ThreadPoolExecutor 
{
+
+        private final AtomicInteger submittedTaskCount = new AtomicInteger();
+
+        private CountingThreadPoolExecutor(int threadCount) {
+            super(threadCount, threadCount, 0L, TimeUnit.MILLISECONDS, new 
LinkedBlockingQueue<>());
+        }
+
+        @Override
+        public void execute(Runnable command) {
+            submittedTaskCount.incrementAndGet();
+            try {
+                super.execute(command);
+            } catch (RuntimeException | Error failure) {
+                submittedTaskCount.decrementAndGet();
+                throw failure;
+            }
+        }
+
+        private int getSubmittedTaskCount() {
+            return submittedTaskCount.get();
+        }
+    }
 }