You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ja...@apache.org on 2025/06/24 03:28:15 UTC
(iotdb) 02/02: Throw CANNOT_FETCH_FI_STATE(722) instead of 301/305 while DN restarting
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch ChangeStatusCode-1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 7cd3ce72cbf3168011ae76f96458eed6525874ab
Author: JackieTien97 <ja...@gmail.com>
AuthorDate: Tue Jun 24 11:12:54 2025 +0800
Throw CANNOT_FETCH_FI_STATE(722) instead of 301/305 while DN restarting
---
.../queryengine/execution/QueryStateMachine.java | 30 +++++++++----------
.../scheduler/FixedRateFragInsStateTracker.java | 34 +++++++++++-----------
2 files changed, 30 insertions(+), 34 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/QueryStateMachine.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/QueryStateMachine.java
index ab9201787a6..581b015eeb6 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/QueryStateMachine.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/QueryStateMachine.java
@@ -26,7 +26,6 @@ import org.apache.iotdb.db.queryengine.plan.execution.QueryExecution;
import com.google.common.util.concurrent.ListenableFuture;
-import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import static com.google.common.base.Preconditions.checkArgument;
@@ -48,19 +47,13 @@ import static org.apache.iotdb.db.queryengine.execution.QueryState.RUNNING;
public class QueryStateMachine {
private final StateMachine<QueryState> queryState;
- // The executor will be used in all the state machines belonged to this query.
- private Executor stateMachineExecutor;
private Throwable failureException;
private TSStatus failureStatus;
public QueryStateMachine(QueryId queryId, ExecutorService executor) {
- this.stateMachineExecutor = executor;
this.queryState =
new StateMachine<>(
- queryId.toString(),
- this.stateMachineExecutor,
- QUEUED,
- QueryState.TERMINAL_INSTANCE_STATES);
+ queryId.toString(), executor, QUEUED, QueryState.TERMINAL_INSTANCE_STATES);
}
public void addStateChangeListener(
@@ -109,9 +102,10 @@ public class QueryStateMachine {
}
public void transitionToCanceled(Throwable throwable, TSStatus failureStatus) {
- this.failureException = throwable;
- this.failureStatus = failureStatus;
- transitionToDoneState(CANCELED);
+ if (transitionToDoneState(CANCELED)) {
+ this.failureException = throwable;
+ this.failureStatus = failureStatus;
+ }
}
public void transitionToAborted() {
@@ -123,20 +117,22 @@ public class QueryStateMachine {
}
public void transitionToFailed(Throwable throwable) {
- this.failureException = throwable;
- transitionToDoneState(FAILED);
+ if (transitionToDoneState(FAILED)) {
+ this.failureException = throwable;
+ }
}
public void transitionToFailed(TSStatus failureStatus) {
- this.failureStatus = failureStatus;
- transitionToDoneState(FAILED);
+ if (transitionToDoneState(FAILED)) {
+ this.failureStatus = failureStatus;
+ }
}
- private void transitionToDoneState(QueryState doneState) {
+ private boolean transitionToDoneState(QueryState doneState) {
requireNonNull(doneState, "doneState is null");
checkArgument(doneState.isDone(), "doneState %s is not a done state", doneState);
- queryState.setIf(doneState, currentState -> !currentState.isDone());
+ return queryState.setIf(doneState, currentState -> !currentState.isDone());
}
public String getFailureMessage() {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/FixedRateFragInsStateTracker.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/FixedRateFragInsStateTracker.java
index f2afe5101ef..b848db6b6d9 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/FixedRateFragInsStateTracker.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/FixedRateFragInsStateTracker.java
@@ -180,8 +180,7 @@ public class FixedRateFragInsStateTracker extends AbstractFragInsStateTracker {
instanceId, instanceInfo.getMessage()),
TSStatusCode.CANNOT_FETCH_FI_STATE.getStatusCode(),
true));
- }
- if (instanceInfo.getState().isFailed()) {
+ } else if (instanceInfo.getState().isFailed()) {
if (instanceInfo.getErrorCode().isPresent()) {
stateMachine.transitionToFailed(
new IoTDBException(
@@ -196,22 +195,23 @@ public class FixedRateFragInsStateTracker extends AbstractFragInsStateTracker {
} else {
stateMachine.transitionToFailed(instanceInfo.getFailureInfoList().get(0).toException());
}
- }
- boolean queryFinished = false;
- List<InstanceStateMetrics> rootInstanceStateMetricsList =
- instanceStateMap.values().stream()
- .filter(instanceStateMetrics -> instanceStateMetrics.isRootInstance)
- .collect(Collectors.toList());
- if (!rootInstanceStateMetricsList.isEmpty()) {
- queryFinished =
- rootInstanceStateMetricsList.stream()
- .allMatch(
- instanceStateMetrics ->
- instanceStateMetrics.lastState == FragmentInstanceState.FINISHED);
- }
+ } else {
+ boolean queryFinished = false;
+ List<InstanceStateMetrics> rootInstanceStateMetricsList =
+ instanceStateMap.values().stream()
+ .filter(instanceStateMetrics -> instanceStateMetrics.isRootInstance)
+ .collect(Collectors.toList());
+ if (!rootInstanceStateMetricsList.isEmpty()) {
+ queryFinished =
+ rootInstanceStateMetricsList.stream()
+ .allMatch(
+ instanceStateMetrics ->
+ instanceStateMetrics.lastState == FragmentInstanceState.FINISHED);
+ }
- if (queryFinished) {
- stateMachine.transitionToFinished();
+ if (queryFinished) {
+ stateMachine.transitionToFinished();
+ }
}
}