This is an automated email from the ASF dual-hosted git repository.
JackieTien97 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 9112791a88c Fix: remove some procedure that should not be included in
lease frame (#18167)
9112791a88c is described below
commit 9112791a88c114227113d083c49d6ad13aaa2b1d
Author: CYB <[email protected]>
AuthorDate: Thu Jul 9 23:03:47 2026 +0800
Fix: remove some procedure that should not be included in lease frame
(#18167)
---
.../impl/schema/DeactivateTemplateProcedure.java | 6 +-
.../procedure/impl/schema/SchemaUtils.java | 40 --------------
.../procedure/impl/schema/SetTTLProcedure.java | 64 +++++++++++++---------
.../impl/schema/SetTemplateProcedure.java | 56 +++++++++----------
.../impl/schema/UnsetTemplateProcedure.java | 46 ++++++++--------
.../procedure/impl/schema/SetTTLProcedureTest.java | 29 ++++++++--
6 files changed, 113 insertions(+), 128 deletions(-)
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeactivateTemplateProcedure.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeactivateTemplateProcedure.java
index 4ca00eb46aa..b2875b7fc41 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeactivateTemplateProcedure.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeactivateTemplateProcedure.java
@@ -35,7 +35,6 @@ import
org.apache.iotdb.confignode.client.async.handlers.DataNodeAsyncRequestCon
import
org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeDeactivateTemplatePlan;
import
org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeEnrichedPlan;
import org.apache.iotdb.confignode.i18n.ProcedureMessages;
-import org.apache.iotdb.confignode.manager.lease.ClusterCachePropagator;
import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
import org.apache.iotdb.confignode.procedure.exception.ProcedureException;
import org.apache.iotdb.confignode.procedure.impl.StateMachineProcedure;
@@ -191,10 +190,7 @@ public class DeactivateTemplateProcedure
new TInvalidateMatchedSchemaCacheReq(timeSeriesPatternTreeBytes),
dataNodeLocationMap);
CnToDnInternalServiceAsyncRequestManager.getInstance()
- .sendAsyncRequest(
- clientHandler,
- ClusterCachePropagator.BROADCAST_RPC_RETRY,
- ClusterCachePropagator.BROADCAST_RPC_TIMEOUT_MS);
+ .sendAsyncRequestWithRetry(clientHandler);
Map<Integer, TSStatus> statusMap = clientHandler.getResponseMap();
for (TSStatus status : statusMap.values()) {
// all dataNodes must clear the related schema cache
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/SchemaUtils.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/SchemaUtils.java
index 26d9c8df391..69a3cec4400 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/SchemaUtils.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/SchemaUtils.java
@@ -48,7 +48,6 @@ import
org.apache.iotdb.mpp.rpc.thrift.TCountPathsUsingTemplateReq;
import org.apache.iotdb.mpp.rpc.thrift.TCountPathsUsingTemplateResp;
import org.apache.iotdb.mpp.rpc.thrift.TInvalidateMatchedSchemaCacheReq;
import org.apache.iotdb.mpp.rpc.thrift.TUpdateTableReq;
-import org.apache.iotdb.mpp.rpc.thrift.TUpdateTemplateReq;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;
@@ -65,7 +64,6 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
-import java.util.function.Supplier;
import java.util.stream.Collectors;
public class SchemaUtils {
@@ -337,17 +335,6 @@ public class SchemaUtils {
return req;
}
- public static Map<Integer, TSStatus> rollbackPreRelease(
- final String database,
- final String tableName,
- final ConfigManager configManager,
- final @Nullable String oldName) {
- return failedOnly(
- broadcastTableUpdate(
- rollbackUpdateTableReq(database, tableName, oldName),
- filterFencedDataNode(configManager)));
- }
-
/**
* Broadcast an INVALIDATE_MATCHED_SCHEMA_CACHE to all DataNodes through
{@link
* ClusterCachePropagator}: proceed once every unreachable DataNode is
provably self-fenced (it
@@ -382,33 +369,6 @@ public class SchemaUtils {
});
}
- /**
- * Broadcast an UPDATE_TEMPLATE to all DataNodes through {@link
ClusterCachePropagator}: proceed
- * once every unreachable DataNode is provably self-fenced (it fails closed
on its template cache
- * and resyncs on recovery), instead of hard-failing on the first
unreachable DataNode. Returns
- * whether it is safe to proceed.
- *
- * <p>The request is rebuilt from {@code requestSupplier} on every attempt:
the propagator may
- * re-broadcast while waiting, and {@code TUpdateTemplateReq}'s binary field
is backed by a {@link
- * ByteBuffer}, so reusing one request could re-send a consumed (empty)
payload.
- */
- public static boolean broadcastTemplateUpdate(
- final ConfigManager configManager, final Supplier<TUpdateTemplateReq>
requestSupplier) {
- return new ClusterCachePropagator(filterFencedDataNode(configManager))
- .propagate(
- targets -> {
- final DataNodeAsyncRequestContext<TUpdateTemplateReq, TSStatus>
clientHandler =
- new DataNodeAsyncRequestContext<>(
- CnToDnAsyncRequestType.UPDATE_TEMPLATE,
requestSupplier.get(), targets);
- CnToDnInternalServiceAsyncRequestManager.getInstance()
- .sendAsyncRequest(
- clientHandler,
- ClusterCachePropagator.BROADCAST_RPC_RETRY,
- ClusterCachePropagator.BROADCAST_RPC_TIMEOUT_MS);
- return clientHandler.getResponseMap();
- });
- }
-
public static TSStatus executeInConsensusLayer(
final ConfigPhysicalPlan plan, final ConfigNodeProcedureEnv env, final
Logger logger) {
TSStatus status;
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/SetTTLProcedure.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/SetTTLProcedure.java
index 57a9b3b1ceb..9ecd4c35b28 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/SetTTLProcedure.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/SetTTLProcedure.java
@@ -34,7 +34,6 @@ import
org.apache.iotdb.confignode.consensus.request.write.database.SetTTLPlan;
import
org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeEnrichedPlan;
import org.apache.iotdb.confignode.i18n.ConfigNodeMessages;
import org.apache.iotdb.confignode.i18n.ProcedureMessages;
-import org.apache.iotdb.confignode.manager.lease.ClusterCachePropagator;
import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
import org.apache.iotdb.confignode.procedure.exception.ProcedureException;
import org.apache.iotdb.confignode.procedure.impl.StateMachineProcedure;
@@ -116,8 +115,13 @@ public class SetTTLProcedure extends
StateMachineProcedure<ConfigNodeProcedureEn
}
void updateDataNodeTTL(final ConfigNodeProcedureEnv env) {
- if (!broadcastTTLAndDecide(
- env, buildSetTTLReq(plan.getPathPattern(), plan.getTTL(),
plan.isDataBase()))) {
+ final Map<Integer, TDataNodeLocation> dataNodeLocationMap =
+
env.getConfigManager().getNodeManager().getRegisteredDataNodeLocations();
+ final DataNodeAsyncRequestContext<TSetTTLReq, TSStatus> clientHandler =
+ sendTTLRequest(
+ dataNodeLocationMap,
+ buildSetTTLReq(plan.getPathPattern(), plan.getTTL(),
plan.isDataBase()));
+ if (hasFailedDataNode(clientHandler)) {
LOGGER.error(ProcedureMessages.FAILED_TO_UPDATE_TTL_CACHE_OF_DATANODE);
setFailure(
new ProcedureException(
@@ -125,17 +129,6 @@ public class SetTTLProcedure extends
StateMachineProcedure<ConfigNodeProcedureEn
}
}
- /**
- * Broadcast the TTL update to all DataNodes and decide whether it is safe
to proceed: proceed
- * once every unreachable DataNode is provably self-fenced (it fails closed
on TTL in compaction
- * and resyncs on recovery) instead of hard-failing on the first unreachable
DataNode.
- * Package-private and overridable for tests.
- */
- boolean broadcastTTLAndDecide(final ConfigNodeProcedureEnv env, final
TSetTTLReq req) {
- return new
ClusterCachePropagator(SchemaUtils.filterFencedDataNode(env.getConfigManager()))
- .propagate(targets -> sendTTLRequest(targets, req).getResponseMap());
- }
-
private void capturePreviousTTLState(final ConfigNodeProcedureEnv env) {
if (previousTTLStateCaptured) {
return;
@@ -165,11 +158,7 @@ public class SetTTLProcedure extends
StateMachineProcedure<ConfigNodeProcedureEn
final Map<Integer, TDataNodeLocation> dataNodeLocationMap, final
TSetTTLReq req) {
final DataNodeAsyncRequestContext<TSetTTLReq, TSStatus> clientHandler =
new DataNodeAsyncRequestContext<>(CnToDnAsyncRequestType.SET_TTL, req,
dataNodeLocationMap);
- CnToDnInternalServiceAsyncRequestManager.getInstance()
- .sendAsyncRequest(
- clientHandler,
- ClusterCachePropagator.BROADCAST_RPC_RETRY,
- ClusterCachePropagator.BROADCAST_RPC_TIMEOUT_MS);
+
CnToDnInternalServiceAsyncRequestManager.getInstance().sendAsyncRequestWithRetry(clientHandler);
return clientHandler;
}
@@ -179,6 +168,19 @@ public class SetTTLProcedure extends
StateMachineProcedure<ConfigNodeProcedureEn
Collections.singletonList(String.join(".", pathPattern)), ttl,
isDataBase);
}
+ private boolean hasFailedDataNode(
+ final DataNodeAsyncRequestContext<TSetTTLReq, TSStatus> clientHandler) {
+ if (!clientHandler.getRequestIndices().isEmpty()) {
+ return true;
+ }
+ for (TSStatus status : clientHandler.getResponseMap().values()) {
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
private long getTTLOrDefault(final ConfigNodeProcedureEnv env, final
String[] pathPattern) {
final long ttl =
env.getConfigManager().getTTLManager().getTTL(pathPattern);
return ttl == TTLCache.NULL_TTL ? TTL_NOT_EXIST : ttl;
@@ -218,20 +220,30 @@ public class SetTTLProcedure extends
StateMachineProcedure<ConfigNodeProcedureEn
}
private void rollbackDataNodeTTL(final ConfigNodeProcedureEnv env) throws
ProcedureException {
- restoreTTLOnDataNodes(env, plan.getPathPattern(), previousTTL);
+ final Map<Integer, TDataNodeLocation> dataNodeLocationMap =
+
env.getConfigManager().getNodeManager().getRegisteredDataNodeLocations();
+ restoreTTLOnDataNodes(dataNodeLocationMap, plan.getPathPattern(),
previousTTL);
if (plan.isDataBase()) {
restoreTTLOnDataNodes(
- env, getDatabaseWildcardPathPattern(plan.getPathPattern()),
previousDatabaseWildcardTTL);
+ dataNodeLocationMap,
+ getDatabaseWildcardPathPattern(plan.getPathPattern()),
+ previousDatabaseWildcardTTL);
}
}
private void restoreTTLOnDataNodes(
- final ConfigNodeProcedureEnv env, final String[] pathPattern, final long
ttl)
+ final Map<Integer, TDataNodeLocation> dataNodeLocationMap,
+ final String[] pathPattern,
+ final long ttl)
throws ProcedureException {
- // Same proceed-past-fenced semantics as the forward update: a down
DataNode must not block
- // rollback (it resyncs TTL on recovery); only a live unacked DataNode
fails it.
- if (!broadcastTTLAndDecide(
- env, buildSetTTLReq(pathPattern, ttl == TTL_NOT_EXIST ?
TTLCache.NULL_TTL : ttl, false))) {
+ if (dataNodeLocationMap.isEmpty()) {
+ return;
+ }
+ final DataNodeAsyncRequestContext<TSetTTLReq, TSStatus> clientHandler =
+ sendTTLRequest(
+ dataNodeLocationMap,
+ buildSetTTLReq(pathPattern, ttl == TTL_NOT_EXIST ?
TTLCache.NULL_TTL : ttl, false));
+ if (hasFailedDataNode(clientHandler)) {
throw new ProcedureException(
new MetadataException(
ProcedureMessages.EXCEPTION_ROLLBACK_DATANODE_TTL_CACHE_FAILED_AF9C7102
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/SetTemplateProcedure.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/SetTemplateProcedure.java
index 5a16b49c1f1..52406f2519c 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/SetTemplateProcedure.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/SetTemplateProcedure.java
@@ -40,7 +40,6 @@ import
org.apache.iotdb.confignode.consensus.request.write.template.PreSetSchema
import
org.apache.iotdb.confignode.consensus.response.template.TemplateInfoResp;
import org.apache.iotdb.confignode.i18n.ConfigNodeMessages;
import org.apache.iotdb.confignode.i18n.ProcedureMessages;
-import org.apache.iotdb.confignode.manager.lease.ClusterCachePropagator;
import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
import org.apache.iotdb.confignode.procedure.exception.ProcedureException;
import org.apache.iotdb.confignode.procedure.impl.StateMachineProcedure;
@@ -216,25 +215,30 @@ public class SetTemplateProcedure
return;
}
- // Proceed once every unreachable DataNode is provably self-fenced (it
fails closed on its
- // template cache and resyncs on recovery) instead of hard-failing on the
first unreachable one.
- if (!SchemaUtils.broadcastTemplateUpdate(
- env.getConfigManager(),
- () -> {
- final TUpdateTemplateReq req = new TUpdateTemplateReq();
-
req.setType(TemplateInternalRPCUpdateType.ADD_TEMPLATE_PRE_SET_INFO.toByte());
- req.setTemplateInfo(
-
TemplateInternalRPCUtil.generateAddTemplateSetInfoBytes(template,
templateSetPath));
- return req;
- })) {
- LOGGER.warn(
- ProcedureMessages.FAILED_TO_SYNC_TEMPLATE_PRE_SET_INFO_ON_PATH_TO,
- templateName,
- templateSetPath,
- "an unreachable DataNode is not provably fenced");
- setFailure(
- new ProcedureException(new
MetadataException(ProcedureMessages.PRE_SET_TEMPLATE_FAILED)));
- return;
+ final TUpdateTemplateReq req = new TUpdateTemplateReq();
+
req.setType(TemplateInternalRPCUpdateType.ADD_TEMPLATE_PRE_SET_INFO.toByte());
+ req.setTemplateInfo(
+ TemplateInternalRPCUtil.generateAddTemplateSetInfoBytes(template,
templateSetPath));
+
+ final Map<Integer, TDataNodeLocation> dataNodeLocationMap =
+
env.getConfigManager().getNodeManager().getRegisteredDataNodeLocations();
+ final DataNodeAsyncRequestContext<TUpdateTemplateReq, TSStatus>
clientHandler =
+ new DataNodeAsyncRequestContext<>(
+ CnToDnAsyncRequestType.UPDATE_TEMPLATE, req, dataNodeLocationMap);
+
CnToDnInternalServiceAsyncRequestManager.getInstance().sendAsyncRequestWithRetry(clientHandler);
+ final Map<Integer, TSStatus> statusMap = clientHandler.getResponseMap();
+ for (final Map.Entry<Integer, TSStatus> entry : statusMap.entrySet()) {
+ if (entry.getValue().getCode() !=
TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ LOGGER.warn(
+ ProcedureMessages.FAILED_TO_SYNC_TEMPLATE_PRE_SET_INFO_ON_PATH_TO,
+ templateName,
+ templateSetPath,
+ dataNodeLocationMap.get(entry.getKey()));
+ setFailure(
+ new ProcedureException(
+ new
MetadataException(ProcedureMessages.PRE_SET_TEMPLATE_FAILED)));
+ return;
+ }
}
setNextState(SetTemplateState.VALIDATE_TIMESERIES_EXISTENCE);
}
@@ -407,11 +411,7 @@ public class SetTemplateProcedure
final DataNodeAsyncRequestContext<TUpdateTemplateReq, TSStatus>
clientHandler =
new DataNodeAsyncRequestContext<>(
CnToDnAsyncRequestType.UPDATE_TEMPLATE, req, dataNodeLocationMap);
- CnToDnInternalServiceAsyncRequestManager.getInstance()
- .sendAsyncRequest(
- clientHandler,
- ClusterCachePropagator.BROADCAST_RPC_RETRY,
- ClusterCachePropagator.BROADCAST_RPC_TIMEOUT_MS);
+
CnToDnInternalServiceAsyncRequestManager.getInstance().sendAsyncRequestWithRetry(clientHandler);
final Map<Integer, TSStatus> statusMap = clientHandler.getResponseMap();
for (final Map.Entry<Integer, TSStatus> entry : statusMap.entrySet()) {
if (entry.getValue().getCode() !=
TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
@@ -523,11 +523,7 @@ public class SetTemplateProcedure
CnToDnAsyncRequestType.UPDATE_TEMPLATE,
invalidateTemplateSetInfoReq,
dataNodeLocationMap);
- CnToDnInternalServiceAsyncRequestManager.getInstance()
- .sendAsyncRequest(
- clientHandler,
- ClusterCachePropagator.BROADCAST_RPC_RETRY,
- ClusterCachePropagator.BROADCAST_RPC_TIMEOUT_MS);
+
CnToDnInternalServiceAsyncRequestManager.getInstance().sendAsyncRequestWithRetry(clientHandler);
final Map<Integer, TSStatus> statusMap = clientHandler.getResponseMap();
for (final Map.Entry<Integer, TSStatus> entry : statusMap.entrySet()) {
// all dataNodes must clear the related template cache
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/UnsetTemplateProcedure.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/UnsetTemplateProcedure.java
index e97af50aa09..1fd7aefb330 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/UnsetTemplateProcedure.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/UnsetTemplateProcedure.java
@@ -31,7 +31,6 @@ import
org.apache.iotdb.confignode.client.async.CnToDnAsyncRequestType;
import
org.apache.iotdb.confignode.client.async.CnToDnInternalServiceAsyncRequestManager;
import
org.apache.iotdb.confignode.client.async.handlers.DataNodeAsyncRequestContext;
import org.apache.iotdb.confignode.i18n.ProcedureMessages;
-import org.apache.iotdb.confignode.manager.lease.ClusterCachePropagator;
import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
import org.apache.iotdb.confignode.procedure.exception.ProcedureException;
import org.apache.iotdb.confignode.procedure.impl.StateMachineProcedure;
@@ -156,24 +155,29 @@ public class UnsetTemplateProcedure
}
private void executeInvalidateCache(final ConfigNodeProcedureEnv env) throws
ProcedureException {
- // Proceed once every unreachable DataNode is provably self-fenced (it
fails closed on its
- // template cache and resyncs on recovery) instead of hard-failing on the
first unreachable one.
- if (!SchemaUtils.broadcastTemplateUpdate(
- env.getConfigManager(),
- () -> {
- final TUpdateTemplateReq invalidateTemplateSetInfoReq = new
TUpdateTemplateReq();
- invalidateTemplateSetInfoReq.setType(
-
TemplateInternalRPCUpdateType.INVALIDATE_TEMPLATE_SET_INFO.toByte());
-
invalidateTemplateSetInfoReq.setTemplateInfo(getInvalidateTemplateSetInfo());
- return invalidateTemplateSetInfoReq;
- })) {
+ final Map<Integer, TDataNodeLocation> dataNodeLocationMap =
+
env.getConfigManager().getNodeManager().getRegisteredDataNodeLocations();
+ final TUpdateTemplateReq invalidateTemplateSetInfoReq = new
TUpdateTemplateReq();
+ invalidateTemplateSetInfoReq.setType(
+ TemplateInternalRPCUpdateType.INVALIDATE_TEMPLATE_SET_INFO.toByte());
+
invalidateTemplateSetInfoReq.setTemplateInfo(getInvalidateTemplateSetInfo());
+ final DataNodeAsyncRequestContext<TUpdateTemplateReq, TSStatus>
clientHandler =
+ new DataNodeAsyncRequestContext<>(
+ CnToDnAsyncRequestType.UPDATE_TEMPLATE,
+ invalidateTemplateSetInfoReq,
+ dataNodeLocationMap);
+
CnToDnInternalServiceAsyncRequestManager.getInstance().sendAsyncRequestWithRetry(clientHandler);
+ final Map<Integer, TSStatus> statusMap = clientHandler.getResponseMap();
+ for (final TSStatus status : statusMap.values()) {
// all dataNodes must clear the related template cache
- LOGGER.error(
-
ProcedureMessages.FAILED_TO_INVALIDATE_TEMPLATE_CACHE_OF_TEMPLATE_SET_ON,
- template.getName(),
- path);
- throw new ProcedureException(
- new
MetadataException(ProcedureMessages.INVALIDATE_TEMPLATE_CACHE_FAILED));
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ LOGGER.error(
+
ProcedureMessages.FAILED_TO_INVALIDATE_TEMPLATE_CACHE_OF_TEMPLATE_SET_ON,
+ template.getName(),
+ path);
+ throw new ProcedureException(
+ new
MetadataException(ProcedureMessages.INVALIDATE_TEMPLATE_CACHE_FAILED));
+ }
}
}
@@ -264,11 +268,7 @@ public class UnsetTemplateProcedure
CnToDnAsyncRequestType.UPDATE_TEMPLATE,
rollbackTemplateSetInfoReq,
dataNodeLocationMap);
- CnToDnInternalServiceAsyncRequestManager.getInstance()
- .sendAsyncRequest(
- clientHandler,
- ClusterCachePropagator.BROADCAST_RPC_RETRY,
- ClusterCachePropagator.BROADCAST_RPC_TIMEOUT_MS);
+
CnToDnInternalServiceAsyncRequestManager.getInstance().sendAsyncRequestWithRetry(clientHandler);
Map<Integer, TSStatus> statusMap = clientHandler.getResponseMap();
for (TSStatus status : statusMap.values()) {
// all dataNodes must clear the related template cache
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/schema/SetTTLProcedureTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/schema/SetTTLProcedureTest.java
index a1813c1642c..cb09c23659c 100644
---
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/schema/SetTTLProcedureTest.java
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/schema/SetTTLProcedureTest.java
@@ -25,6 +25,8 @@ import org.apache.iotdb.common.rpc.thrift.TSetTTLReq;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.schema.ttl.TTLCache;
+import org.apache.iotdb.confignode.client.async.CnToDnAsyncRequestType;
+import
org.apache.iotdb.confignode.client.async.handlers.DataNodeAsyncRequestContext;
import org.apache.iotdb.confignode.consensus.request.write.database.SetTTLPlan;
import org.apache.iotdb.confignode.manager.ConfigManager;
import org.apache.iotdb.confignode.manager.TTLManager;
@@ -353,11 +355,30 @@ public class SetTTLProcedureTest {
}
@Override
- boolean broadcastTTLAndDecide(final ConfigNodeProcedureEnv env, final
TSetTTLReq req) {
+ DataNodeAsyncRequestContext<TSetTTLReq, TSStatus> sendTTLRequest(
+ final Map<Integer, TDataNodeLocation> dataNodeLocationMap, final
TSetTTLReq req) {
requests.add(copyRequest(req));
- // Simulate a live, un-acked DataNode on the first broadcast: the
propagator verdict is FAIL
- // (which triggers rollback). Later broadcasts (the rollback restore)
proceed.
- return !(failFirstDataNodeUpdate && requestCount++ == 0);
+
+ final DataNodeAsyncRequestContext<TSetTTLReq, TSStatus> clientHandler =
+ new DataNodeAsyncRequestContext<>(
+ CnToDnAsyncRequestType.SET_TTL, copyRequest(req),
dataNodeLocationMap);
+ final List<Integer> requestIds = new
ArrayList<>(clientHandler.getNodeLocationMap().keySet());
+ final boolean shouldFail = failFirstDataNodeUpdate && requestCount++ ==
0;
+
+ for (Integer requestId : requestIds) {
+ clientHandler
+ .getResponseMap()
+ .put(
+ requestId,
+ new TSStatus(
+ shouldFail
+ ? TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode()
+ : TSStatusCode.SUCCESS_STATUS.getStatusCode()));
+ if (!shouldFail) {
+ clientHandler.getNodeLocationMap().remove(requestId);
+ }
+ }
+ return clientHandler;
}
private SetTTLPlan copyPlan(final SetTTLPlan plan) {
