(shardingsphere) branch master updated: Throw instead of returning partial results when ExecutorEngine is interrupted (#39400)

[zh...@apache.org]

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

terrymanu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shardingsphere.git


The following commit(s) were added to refs/heads/master by this push:
     new 4d3d7267665 Throw instead of returning partial results when ExecutorEngine is interrupted (#39400)
4d3d7267665 is described below

commit 4d3d7267665d9da9e10fce8fca689e603e99e990
Author: ZIHAN DAI <99...@users.noreply.github.com>
AuthorDate: Tue Aug 11 12:53:08 2026 +1000

    Throw instead of returning partial results when ExecutorEngine is interrupted (#39400)
    
    getGroupResults restored the interrupt flag and then continued the loop, so
    every remaining Future.get() threw immediately and the method returned a result
    list silently missing one entry per interrupted group, reported as success.
    
    The sibling ExecutionException handler two lines below already terminates via
    throwException. Do the same after re-interrupting.
    
    Fixes #39399
---
 .../infra/executor/kernel/ExecutorEngine.java      |  1 +
 .../infra/executor/kernel/ExecutorEngineTest.java  | 36 ++++++++++++++++++++++
 2 files changed, 37 insertions(+)

diff --git a/infra/executor/src/main/java/org/apache/shardingsphere/infra/executor/kernel/ExecutorEngine.java b/infra/executor/src/main/java/org/apache/shardingsphere/infra/executor/kernel/ExecutorEngine.java
index a659d73f502..55d99d3a1cb 100644
--- a/infra/executor/src/main/java/org/apache/shardingsphere/infra/executor/kernel/ExecutorEngine.java
+++ b/infra/executor/src/main/java/org/apache/shardingsphere/infra/executor/kernel/ExecutorEngine.java
@@ -118,6 +118,7 @@ public final class ExecutorEngine implements AutoCloseable {
                 result.addAll(each.get());
             } catch (final InterruptedException ex) {
                 Thread.currentThread().interrupt();
+                return throwException(ex);
             } catch (final ExecutionException ex) {
                 return throwException(ex);
             }
diff --git a/infra/executor/src/test/java/org/apache/shardingsphere/infra/executor/kernel/ExecutorEngineTest.java b/infra/executor/src/test/java/org/apache/shardingsphere/infra/executor/kernel/ExecutorEngineTest.java
index 236ab235796..6d62ab9fab2 100644
--- a/infra/executor/src/test/java/org/apache/shardingsphere/infra/executor/kernel/ExecutorEngineTest.java
+++ b/infra/executor/src/test/java/org/apache/shardingsphere/infra/executor/kernel/ExecutorEngineTest.java
@@ -25,14 +25,20 @@ import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
+import org.apache.shardingsphere.infra.executor.kernel.model.ExecutorCallback;
+import org.apache.shardingsphere.infra.exception.generic.UnknownSQLException;
+
 import java.sql.SQLException;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.Mockito.mock;
 
 class ExecutorEngineTest {
@@ -88,4 +94,34 @@ class ExecutorEngineTest {
         latch.await();
         assertThat(actual.size(), is(4));
     }
+    
+    @Test
+    void assertParallelExecuteThrowsWhenInterrupted() throws InterruptedException {
+        CountDownLatch neverReleased = new CountDownLatch(1);
+        CountDownLatch asyncStarted = new CountDownLatch(1);
+        // The trunk thread interrupts itself while the async group is still running, so the
+        // Future.get() that follows throws InterruptedException. A future that has already
+        // completed hands back its value even with the flag set, hence the block.
+        ExecutorCallback<Object, String> interruptingFirstCallback = (inputs, isTrunkThread, processId) -> {
+            Thread.currentThread().interrupt();
+            return Collections.singletonList("succeed");
+        };
+        ExecutorCallback<Object, String> blockingCallback = (inputs, isTrunkThread, processId) -> {
+            asyncStarted.countDown();
+            try {
+                neverReleased.await(30L, TimeUnit.SECONDS);
+            } catch (final InterruptedException ignored) {
+                Thread.currentThread().interrupt();
+            }
+            return Collections.singletonList("succeed");
+        };
+        try {
+            assertThrows(UnknownSQLException.class,
+                    () -> executorEngine.execute(executionGroupContext, interruptingFirstCallback, blockingCallback, false));
+        } finally {
+            Thread.interrupted();
+            neverReleased.countDown();
+            asyncStarted.await(30L, TimeUnit.SECONDS);
+        }
+    }
 }