This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new e86ef39d699 Fix repeated RPC dispatch reusing a released
FragmentInstanceContext (NPE) (#17794) (#18235)
e86ef39d699 is described below
commit e86ef39d6992cdb7f3484fffd2f7284ca7ac911e
Author: Jackie Tien <[email protected]>
AuthorDate: Fri Jul 17 18:48:23 2026 +0800
Fix repeated RPC dispatch reusing a released FragmentInstanceContext (NPE)
(#17794) (#18235)
---
.../java/org/apache/iotdb/rpc/TSStatusCode.java | 1 +
.../execution/executor/RegionReadExecutor.java | 23 ++++++-
.../fragment/FragmentInstanceManager.java | 77 ++++++++++++++++------
.../scheduler/FragmentInstanceDispatcherImpl.java | 15 +++++
.../apache/iotdb/db/utils/ErrorHandlingUtils.java | 2 +-
.../execution/executor/RegionReadExecutorTest.java | 46 +++++++++++++
6 files changed, 141 insertions(+), 23 deletions(-)
diff --git
a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
index e2855ed0a50..af4f4727c7a 100644
---
a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
+++
b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
@@ -126,6 +126,7 @@ public enum TSStatusCode {
QUERY_TIMEOUT(720),
PLAN_FAILED_NETWORK_PARTITION(721),
CANNOT_FETCH_FI_STATE(722),
+ REPEATED_RPC_CALL(723),
// Arithmetic
NUMERIC_VALUE_OUT_OF_RANGE(750),
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/executor/RegionReadExecutor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/executor/RegionReadExecutor.java
index 7772c4ceaee..2d69dfd4049 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/executor/RegionReadExecutor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/executor/RegionReadExecutor.java
@@ -22,6 +22,7 @@ package org.apache.iotdb.db.queryengine.execution.executor;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.consensus.ConsensusGroupId;
import org.apache.iotdb.commons.consensus.DataRegionId;
+import org.apache.iotdb.commons.exception.IoTDBRuntimeException;
import org.apache.iotdb.commons.utils.ErrorHandlingCommonUtils;
import org.apache.iotdb.commons.utils.StatusUtils;
import org.apache.iotdb.commons.utils.TestOnly;
@@ -124,6 +125,13 @@ public class RegionReadExecutor {
|| t instanceof InterruptedException) {
resp.setReadNeedRetry(true);
resp.setStatus(new
TSStatus(TSStatusCode.RATIS_READ_UNAVAILABLE.getStatusCode()));
+ } else if (t instanceof IoTDBRuntimeException) {
+ // Carry the original status code (e.g. REPEATED_RPC_CALL) back to the
dispatcher so it is
+ // not downgraded to EXECUTE_STATEMENT_ERROR; needRetryHelper decides
retryability.
+ TSStatus status = new TSStatus(((IoTDBRuntimeException)
t).getErrorCode());
+ status.setMessage(t.getMessage());
+ resp.setStatus(status);
+ resp.setReadNeedRetry(StatusUtils.needRetryHelper(status));
}
return resp;
}
@@ -140,8 +148,19 @@ public class RegionReadExecutor {
return RegionExecutionResult.create(!info.getState().isFailed(),
info.getMessage(), null);
} catch (Throwable t) {
LOGGER.error("Execute FragmentInstance in QueryExecutor failed.", t);
- return RegionExecutionResult.create(
- false, String.format(ERROR_MSG_FORMAT, t.getMessage()), null);
+ RegionExecutionResult resp =
+ RegionExecutionResult.create(
+ false, String.format(ERROR_MSG_FORMAT, t.getMessage()), null);
+ Throwable rootCause = ErrorHandlingCommonUtils.getRootCause(t);
+ if (rootCause instanceof IoTDBRuntimeException) {
+ // Carry the original status code (e.g. REPEATED_RPC_CALL) back to the
dispatcher so it is
+ // not downgraded to EXECUTE_STATEMENT_ERROR; needRetryHelper decides
retryability.
+ TSStatus status = new TSStatus(((IoTDBRuntimeException)
rootCause).getErrorCode());
+ status.setMessage(rootCause.getMessage());
+ resp.setStatus(status);
+ resp.setReadNeedRetry(StatusUtils.needRetryHelper(status));
+ }
+ return resp;
}
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceManager.java
index 93b562444bf..2ed821fe521 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceManager.java
@@ -25,6 +25,7 @@ import org.apache.iotdb.commons.concurrent.ThreadName;
import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil;
import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.exception.IoTDBException;
+import org.apache.iotdb.commons.exception.IoTDBRuntimeException;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.query.QueryTimeoutRuntimeException;
import org.apache.iotdb.db.queryengine.common.FragmentInstanceId;
@@ -45,6 +46,7 @@ import
org.apache.iotdb.db.schemaengine.schemaregion.ISchemaRegion;
import org.apache.iotdb.db.storageengine.dataregion.IDataRegionForQuery;
import org.apache.iotdb.db.utils.SetThreadName;
import org.apache.iotdb.mpp.rpc.thrift.TFetchFragmentInstanceStatisticsResp;
+import org.apache.iotdb.rpc.TSStatusCode;
import io.airlift.units.Duration;
import org.slf4j.Logger;
@@ -143,22 +145,29 @@ public class FragmentInstanceManager {
FragmentInstanceStateMachine stateMachine =
new FragmentInstanceStateMachine(instanceId,
instanceNotificationExecutor);
- int dataNodeFINum = instance.getDataNodeFINum();
- DataNodeQueryContext dataNodeQueryContext =
- getOrCreateDataNodeQueryContext(instanceId.getQueryId(),
dataNodeFINum);
-
+ boolean[] contextCreated = new boolean[] {false};
+ DataNodeQueryContext[] dataNodeQueryContexts = new
DataNodeQueryContext[1];
FragmentInstanceContext context =
instanceContext.computeIfAbsent(
instanceId,
- fragmentInstanceId ->
- createFragmentInstanceContext(
- fragmentInstanceId,
- stateMachine,
- instance.getSessionInfo(),
- dataRegion,
- instance.getGlobalTimePredicate(),
- dataNodeQueryContextMap,
- instance.isDebug()));
+ fragmentInstanceId -> {
+ contextCreated[0] = true;
+ // Only ensure the DataNodeQueryContext when we
actually create the
+ // FragmentInstanceContext, so the repeated-dispatch
path (which rejects
+ // without creating a context) does not leak a
context entry.
+ dataNodeQueryContexts[0] =
+ getOrCreateDataNodeQueryContext(
+ instanceId.getQueryId(),
instance.getDataNodeFINum());
+ return createFragmentInstanceContext(
+ fragmentInstanceId,
+ stateMachine,
+ instance.getSessionInfo(),
+ dataRegion,
+ instance.getGlobalTimePredicate(),
+ dataNodeQueryContextMap,
+ instance.isDebug());
+ });
+ rejectIfRepeatedDispatch(contextCreated[0], instanceId);
context.setHighestPriority(instance.isHighestPriority());
try {
@@ -167,7 +176,7 @@ public class FragmentInstanceManager {
instance.getFragment().getPlanNodeTree(),
instance.getFragment().getTypeProvider(),
context,
- dataNodeQueryContext);
+ dataNodeQueryContexts[0]);
List<IDriver> drivers = new ArrayList<>();
driverFactories.forEach(factory ->
drivers.add(factory.createDriver()));
@@ -231,6 +240,30 @@ public class FragmentInstanceManager {
}
}
+ /**
+ * If {@code instanceContext.computeIfAbsent} returned an existing {@link
FragmentInstanceContext}
+ * for this {@code instanceId} (i.e. {@code contextCreated} is false), the
same FragmentInstance
+ * has been dispatched before (e.g. an RPC retry in {@code
+ * FragmentInstanceDispatcherImpl#dispatchRemote}). The previous execution
may have already
+ * released its resources (dataRegion == null), so reusing this cached
context would run a fresh
+ * driver against a released context and trigger an NPE. Reject the
duplicated dispatch with
+ * REPEATED_RPC_CALL instead of reusing it.
+ *
+ * <p>This must be called before the planning try block on purpose, so it
propagates up
+ * (RegionReadExecutor carries the status code) without touching the first
execution's cached
+ * resources.
+ */
+ private static void rejectIfRepeatedDispatch(
+ boolean contextCreated, FragmentInstanceId instanceId) {
+ if (!contextCreated) {
+ throw new IoTDBRuntimeException(
+ String.format(
+ "Repeated RPC call detected for FragmentInstance %s, reject the
duplicated dispatch.",
+ instanceId.getFullId()),
+ TSStatusCode.REPEATED_RPC_CALL.getStatusCode());
+ }
+ }
+
private void clearFIRelatedResources(FragmentInstanceId instanceId) {
// close and remove all the handles of the fragment instance
exchangeManager.forceDeregisterFragmentInstance(instanceId.toThrift());
@@ -255,15 +288,19 @@ public class FragmentInstanceManager {
FragmentInstanceStateMachine stateMachine =
new FragmentInstanceStateMachine(instanceId,
instanceNotificationExecutor);
+ boolean[] contextCreated = new boolean[] {false};
FragmentInstanceContext context =
instanceContext.computeIfAbsent(
instanceId,
- fragmentInstanceId ->
- createFragmentInstanceContext(
- fragmentInstanceId,
- stateMachine,
- instance.getSessionInfo(),
- instance.isDebug()));
+ fragmentInstanceId -> {
+ contextCreated[0] = true;
+ return createFragmentInstanceContext(
+ fragmentInstanceId,
+ stateMachine,
+ instance.getSessionInfo(),
+ instance.isDebug());
+ });
+ rejectIfRepeatedDispatch(contextCreated[0], instanceId);
context.setHighestPriority(instance.isHighestPriority());
try {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/FragmentInstanceDispatcherImpl.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/FragmentInstanceDispatcherImpl.java
index 3312444f882..a95b4f5e600 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/FragmentInstanceDispatcherImpl.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/FragmentInstanceDispatcherImpl.java
@@ -34,6 +34,7 @@ import
org.apache.iotdb.consensus.exception.ConsensusGroupNotExistException;
import org.apache.iotdb.consensus.exception.RatisReadUnavailableException;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.mpp.FragmentInstanceDispatchException;
+import org.apache.iotdb.db.exception.query.QueryTimeoutRuntimeException;
import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
import
org.apache.iotdb.db.queryengine.execution.executor.RegionExecutionResult;
import org.apache.iotdb.db.queryengine.execution.executor.RegionReadExecutor;
@@ -550,6 +551,20 @@ public class FragmentInstanceDispatcherImpl implements
IFragInstanceDispatcher {
"can't execute request on node {}, error msg is {}, and we try to
reconnect this node.",
endPoint,
ExceptionUtils.getRootCause(e).toString());
+ // If the query has already timed out, do not retry. Re-dispatching the
same FragmentInstance
+ // may cause it to be executed twice on the remote node (see the
REPEATED_RPC_CALL handling in
+ // FragmentInstanceManager), so we fail fast with a timeout status
instead.
+ long currentTime = System.currentTimeMillis();
+ if (currentTime - queryContext.getStartTime() >=
queryContext.getTimeOut()) {
+ throw new FragmentInstanceDispatchException(
+ RpcUtils.getStatus(
+ TSStatusCode.QUERY_TIMEOUT,
+ String.format(
+
QueryTimeoutRuntimeException.QUERY_TIMEOUT_EXCEPTION_MESSAGE,
+ queryContext.getStartTime(),
+ queryContext.getStartTime() + queryContext.getTimeOut(),
+ currentTime)));
+ }
// we just retry once to clear stale connection for a restart node.
try {
dispatchRemoteHelper(instance, endPoint);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/ErrorHandlingUtils.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/ErrorHandlingUtils.java
index 765f19c31bf..8da705ea6da 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/ErrorHandlingUtils.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/ErrorHandlingUtils.java
@@ -159,7 +159,7 @@ public class ErrorHandlingUtils {
Throwable t = e instanceof ExecutionException ? e.getCause() : e;
if (t instanceof QueryTimeoutRuntimeException) {
- return RpcUtils.getStatus(TSStatusCode.INTERNAL_REQUEST_TIME_OUT,
rootCause.getMessage());
+ return RpcUtils.getStatus(TSStatusCode.QUERY_TIMEOUT,
rootCause.getMessage());
} else if (t instanceof ParseCancellationException) {
return RpcUtils.getStatus(
TSStatusCode.SQL_PARSE_ERROR, INFO_PARSING_SQL_ERROR +
rootCause.getMessage());
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/executor/RegionReadExecutorTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/executor/RegionReadExecutorTest.java
index 016fe90b460..b0b5fb805a3 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/executor/RegionReadExecutorTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/executor/RegionReadExecutorTest.java
@@ -22,6 +22,7 @@ package org.apache.iotdb.db.queryengine.execution.executor;
import org.apache.iotdb.commons.consensus.ConsensusGroupId;
import org.apache.iotdb.commons.consensus.DataRegionId;
import org.apache.iotdb.commons.consensus.SchemaRegionId;
+import org.apache.iotdb.commons.exception.IoTDBRuntimeException;
import org.apache.iotdb.consensus.IConsensus;
import org.apache.iotdb.consensus.exception.ConsensusException;
import org.apache.iotdb.db.queryengine.common.FragmentInstanceId;
@@ -30,6 +31,7 @@ import
org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceInfo;
import
org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceManager;
import org.apache.iotdb.db.queryengine.plan.planner.plan.FragmentInstance;
import org.apache.iotdb.db.storageengine.dataregion.VirtualDataRegion;
+import org.apache.iotdb.rpc.TSStatusCode;
import org.junit.Test;
import org.mockito.Mockito;
@@ -161,6 +163,50 @@ public class RegionReadExecutorTest {
assertEquals(String.format(ERROR_MSG_FORMAT, "schema-exception"),
res.getMessage());
}
+ @Test
+ public void testRepeatedRpcCall() throws ConsensusException {
+ // A repeated RPC dispatch of the same FragmentInstance makes
FragmentInstanceManager throw
+ // IoTDBRuntimeException(REPEATED_RPC_CALL). RegionReadExecutor must carry
that status code back
+ // (instead of dropping it, which would downgrade it to
EXECUTE_STATEMENT_ERROR) and mark the
+ // result as non-retryable.
+ ConsensusGroupId dataRegionGroupId = new DataRegionId(1);
+ FragmentInstanceId fragmentInstanceId =
+ new FragmentInstanceId(new PlanFragmentId(MOCK_QUERY_ID, 0), "0");
+ FragmentInstance fragmentInstance = Mockito.mock(FragmentInstance.class);
+ Mockito.when(fragmentInstance.getId()).thenReturn(fragmentInstanceId);
+
+ IConsensus dataRegionConsensus = Mockito.mock(IConsensus.class);
+ IConsensus schemaRegionConsensus = Mockito.mock(IConsensus.class);
+ FragmentInstanceManager fragmentInstanceManager =
Mockito.mock(FragmentInstanceManager.class);
+
+ RegionReadExecutor executor =
+ new RegionReadExecutor(dataRegionConsensus, schemaRegionConsensus,
fragmentInstanceManager);
+
+ // consensus read path (covers both data and schema region queries)
+ Mockito.when(dataRegionConsensus.read(dataRegionGroupId, fragmentInstance))
+ .thenThrow(
+ new IoTDBRuntimeException("repeated",
TSStatusCode.REPEATED_RPC_CALL.getStatusCode()));
+
+ RegionExecutionResult res = executor.execute(dataRegionGroupId,
fragmentInstance);
+
+ assertFalse(res.isAccepted());
+ assertEquals(TSStatusCode.REPEATED_RPC_CALL.getStatusCode(),
res.getStatus().getCode());
+ assertFalse(res.isReadNeedRetry());
+
+ // VirtualDataRegion path (FI executed directly through
FragmentInstanceManager)
+ Mockito.when(
+ fragmentInstanceManager.execDataQueryFragmentInstance(
+ fragmentInstance, VirtualDataRegion.getInstance()))
+ .thenThrow(
+ new IoTDBRuntimeException("repeated",
TSStatusCode.REPEATED_RPC_CALL.getStatusCode()));
+
+ res = executor.execute(fragmentInstance);
+
+ assertFalse(res.isAccepted());
+ assertEquals(TSStatusCode.REPEATED_RPC_CALL.getStatusCode(),
res.getStatus().getCode());
+ assertFalse(res.isReadNeedRetry());
+ }
+
@Test
public void testExceptionHappened() throws ConsensusException {
