You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by xi...@apache.org on 2022/04/21 09:39:39 UTC
[iotdb] 01/01: Add block cancel when GetBlockTask throws exception
This is an automated email from the ASF dual-hosted git repository.
xingtanzjr pushed a commit to branch xingtanzjr/fix_sourceHandle
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 3619f293e0e25346fcd39d4eb33dd234fc85545b
Author: Jinrui.Zhang <xi...@gmail.com>
AuthorDate: Thu Apr 21 17:39:23 2022 +0800
Add block cancel when GetBlockTask throws exception
---
.../src/main/java/org/apache/iotdb/db/mpp/buffer/SourceHandle.java | 6 ++++++
.../main/java/org/apache/iotdb/db/mpp/execution/QueryExecution.java | 3 ++-
2 files changed, 8 insertions(+), 1 deletion(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/buffer/SourceHandle.java b/server/src/main/java/org/apache/iotdb/db/mpp/buffer/SourceHandle.java
index 37a5bce1b0..f7076c278d 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/buffer/SourceHandle.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/buffer/SourceHandle.java
@@ -356,6 +356,12 @@ public class SourceHandle implements ISourceHandle {
localMemoryManager
.getQueryPool()
.free(localFragmentInstanceId.getQueryId(), reservedBytes);
+ // That GetDataBlocksTask throws exception means some TsBlock cannot be
+ // fetched permanently. Someone maybe waiting the SourceHandle to be
+ // unblocked after fetching data. So the blocked should be released.
+ if (blocked != null && !blocked.isDone()) {
+ blocked.cancel(true);
+ }
}
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/QueryExecution.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/QueryExecution.java
index f362a48a72..ad9a1c170a 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/QueryExecution.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/QueryExecution.java
@@ -50,6 +50,7 @@ import java.io.IOException;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.CancellationException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
@@ -210,7 +211,7 @@ public class QueryExecution implements IQueryExecution {
}
return resultHandle.receive();
- } catch (ExecutionException | IOException e) {
+ } catch (ExecutionException | IOException | CancellationException e) {
stateMachine.transitionToFailed(e);
throwIfUnchecked(e.getCause());
throw new RuntimeException(e.getCause());