[iotdb] branch master updated: Add Thread re-interrupt when catching exception (#5604)

[xi...@apache.org]

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

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


The following commit(s) were added to refs/heads/master by this push:
     new a019934b42 Add Thread re-interrupt when catching exception (#5604)
a019934b42 is described below

commit a019934b42b352b21b4ee9033878bfff232f5de0
Author: Zhang.Jinrui <xi...@gmail.com>
AuthorDate: Wed Apr 20 14:54:54 2022 +0800

    Add Thread re-interrupt when catching exception (#5604)
---
 .../db/mpp/execution/config/ConfigExecution.java   |  1 +
 .../mpp/execution/scheduler/ClusterScheduler.java  |  1 +
 .../db/mpp/execution/ConfigExecutionTest.java      | 45 ++++++++++++----------
 3 files changed, 27 insertions(+), 20 deletions(-)

diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/config/ConfigExecution.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/config/ConfigExecution.java
index b134d4896b..fe63991dfc 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/config/ConfigExecution.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/config/ConfigExecution.java
@@ -91,6 +91,7 @@ public class ConfigExecution implements IQueryExecution {
           },
           executor);
     } catch (Throwable e) {
+      Thread.currentThread().interrupt();
       fail(e);
     }
   }
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/scheduler/ClusterScheduler.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/scheduler/ClusterScheduler.java
index 7fa8d3c0b6..18c9b47367 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/scheduler/ClusterScheduler.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/scheduler/ClusterScheduler.java
@@ -96,6 +96,7 @@ public class ClusterScheduler implements IScheduler {
       }
     } catch (InterruptedException | ExecutionException e) {
       // If the dispatch failed, we make the QueryState as failed, and return.
+      Thread.currentThread().interrupt();
       stateMachine.transitionToFailed(e);
       return;
     }
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/execution/ConfigExecutionTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/execution/ConfigExecutionTest.java
index dcc1d2dff6..7ee608d7a0 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/execution/ConfigExecutionTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/execution/ConfigExecutionTest.java
@@ -61,26 +61,6 @@ public class ConfigExecutionTest {
     Assert.assertEquals(TSStatusCode.QUERY_PROCESS_ERROR.getStatusCode(), result.status.code);
   }
 
-  @Test
-  public void exceptionAfterInvokeGetStatusTest() throws InterruptedException {
-    IConfigTask task =
-        () -> {
-          throw new RuntimeException("task throw exception when executing");
-        };
-    ConfigExecution execution =
-        new ConfigExecution(genMPPQueryContext(), null, getExecutor(), task);
-    Thread resultThread =
-        new Thread(
-            () -> {
-              ExecutionResult result = execution.getStatus();
-              Assert.assertEquals(
-                  TSStatusCode.QUERY_PROCESS_ERROR.getStatusCode(), result.status.code);
-            });
-    resultThread.start();
-    execution.start();
-    resultThread.join();
-  }
-
   @Test
   public void configTaskCancelledTest() throws InterruptedException {
     SettableFuture<Void> taskResult = SettableFuture.create();
@@ -113,6 +93,31 @@ public class ConfigExecutionTest {
     resultThread.join();
   }
 
+  @Test
+  public void exceptionAfterInvokeGetStatusTest() {
+    IConfigTask task =
+        () -> {
+          throw new RuntimeException("task throw exception when executing");
+        };
+    ConfigExecution execution =
+        new ConfigExecution(genMPPQueryContext(), null, getExecutor(), task);
+    Thread resultThread =
+        new Thread(
+            () -> {
+              ExecutionResult result = execution.getStatus();
+              Assert.assertEquals(
+                  TSStatusCode.QUERY_PROCESS_ERROR.getStatusCode(), result.status.code);
+            });
+    resultThread.start();
+    execution.start();
+    try {
+      resultThread.join();
+      Assert.fail("InterruptedException should be threw here");
+    } catch (InterruptedException e) {
+      execution.stop();
+    }
+  }
+
   private MPPQueryContext genMPPQueryContext() {
     MPPQueryContext context = new MPPQueryContext(new QueryId("query1"));
     context.setQueryType(QueryType.WRITE);