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 9fa3c49d6d098..ab4913da04d81 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.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 @@ private void invalidateCache(final ConfigNodeProcedureEnv env) { new TInvalidateMatchedSchemaCacheReq(timeSeriesPatternTreeBytes), dataNodeLocationMap); CnToDnInternalServiceAsyncRequestManager.getInstance() - .sendAsyncRequest( - clientHandler, - ClusterCachePropagator.BROADCAST_RPC_RETRY, - ClusterCachePropagator.BROADCAST_RPC_TIMEOUT_MS); + .sendAsyncRequestWithRetry(clientHandler); Map 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 26d9c8df3914b..69a3cec44005d 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.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.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 static TUpdateTableReq rollbackUpdateTableReq( return req; } - public static Map 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 static boolean invalidateMatchedSchemaCache( }); } - /** - * 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. - * - *

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 requestSupplier) { - return new ClusterCachePropagator(filterFencedDataNode(configManager)) - .propagate( - targets -> { - final DataNodeAsyncRequestContext 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 57a9b3b1ceb04..9ecd4c35b2809 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.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 @@ void setConfigNodeTTL(final ConfigNodeProcedureEnv env) { } void updateDataNodeTTL(final ConfigNodeProcedureEnv env) { - if (!broadcastTTLAndDecide( - env, buildSetTTLReq(plan.getPathPattern(), plan.getTTL(), plan.isDataBase()))) { + final Map dataNodeLocationMap = + env.getConfigManager().getNodeManager().getRegisteredDataNodeLocations(); + final DataNodeAsyncRequestContext 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 @@ env, buildSetTTLReq(plan.getPathPattern(), plan.getTTL(), plan.isDataBase()))) { } } - /** - * 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 @@ DataNodeAsyncRequestContext sendTTLRequest( final Map dataNodeLocationMap, final TSetTTLReq req) { final DataNodeAsyncRequestContext 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 @@ private TSetTTLReq buildSetTTLReq( Collections.singletonList(String.join(".", pathPattern)), ttl, isDataBase); } + private boolean hasFailedDataNode( + final DataNodeAsyncRequestContext 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 @@ private void restoreTTLOnConfigNode( } private void rollbackDataNodeTTL(final ConfigNodeProcedureEnv env) throws ProcedureException { - restoreTTLOnDataNodes(env, plan.getPathPattern(), previousTTL); + final Map 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 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 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 04ab1d3496323..964f907c850b1 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.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 @@ private void preReleaseTemplate(final ConfigNodeProcedureEnv env) { 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 dataNodeLocationMap = + env.getConfigManager().getNodeManager().getRegisteredDataNodeLocations(); + final DataNodeAsyncRequestContext clientHandler = + new DataNodeAsyncRequestContext<>( + CnToDnAsyncRequestType.UPDATE_TEMPLATE, req, dataNodeLocationMap); + CnToDnInternalServiceAsyncRequestManager.getInstance().sendAsyncRequestWithRetry(clientHandler); + final Map statusMap = clientHandler.getResponseMap(); + for (final Map.Entry 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 @@ private void commitReleaseTemplate(final ConfigNodeProcedureEnv env) { final DataNodeAsyncRequestContext 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 statusMap = clientHandler.getResponseMap(); for (final Map.Entry entry : statusMap.entrySet()) { if (entry.getValue().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { @@ -523,11 +523,7 @@ private void rollbackPreRelease(final ConfigNodeProcedureEnv env) { CnToDnAsyncRequestType.UPDATE_TEMPLATE, invalidateTemplateSetInfoReq, dataNodeLocationMap); - CnToDnInternalServiceAsyncRequestManager.getInstance() - .sendAsyncRequest( - clientHandler, - ClusterCachePropagator.BROADCAST_RPC_RETRY, - ClusterCachePropagator.BROADCAST_RPC_TIMEOUT_MS); + CnToDnInternalServiceAsyncRequestManager.getInstance().sendAsyncRequestWithRetry(clientHandler); final Map statusMap = clientHandler.getResponseMap(); for (final Map.Entry 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 e97af50aa09f4..1fd7aefb33065 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.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 @@ private void invalidateCache(final ConfigNodeProcedureEnv env) { } 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 dataNodeLocationMap = + env.getConfigManager().getNodeManager().getRegisteredDataNodeLocations(); + final TUpdateTemplateReq invalidateTemplateSetInfoReq = new TUpdateTemplateReq(); + invalidateTemplateSetInfoReq.setType( + TemplateInternalRPCUpdateType.INVALIDATE_TEMPLATE_SET_INFO.toByte()); + invalidateTemplateSetInfoReq.setTemplateInfo(getInvalidateTemplateSetInfo()); + final DataNodeAsyncRequestContext clientHandler = + new DataNodeAsyncRequestContext<>( + CnToDnAsyncRequestType.UPDATE_TEMPLATE, + invalidateTemplateSetInfoReq, + dataNodeLocationMap); + CnToDnInternalServiceAsyncRequestManager.getInstance().sendAsyncRequestWithRetry(clientHandler); + final Map 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 @@ private void executeRollbackInvalidateCache(ConfigNodeProcedureEnv env) CnToDnAsyncRequestType.UPDATE_TEMPLATE, rollbackTemplateSetInfoReq, dataNodeLocationMap); - CnToDnInternalServiceAsyncRequestManager.getInstance() - .sendAsyncRequest( - clientHandler, - ClusterCachePropagator.BROADCAST_RPC_RETRY, - ClusterCachePropagator.BROADCAST_RPC_TIMEOUT_MS); + CnToDnInternalServiceAsyncRequestManager.getInstance().sendAsyncRequestWithRetry(clientHandler); Map 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 a1813c1642cca..cb09c23659c39 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.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 @@ TSStatus writeConfigNodePlan(final ConfigNodeProcedureEnv env, final SetTTLPlan } @Override - boolean broadcastTTLAndDecide(final ConfigNodeProcedureEnv env, final TSetTTLReq req) { + DataNodeAsyncRequestContext sendTTLRequest( + final Map 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 clientHandler = + new DataNodeAsyncRequestContext<>( + CnToDnAsyncRequestType.SET_TTL, copyRequest(req), dataNodeLocationMap); + final List 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) {