Skip to content

Fix: remove some procedure that should not be included in lease frame #18167

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
JackieTien97 merged 1 commit into from
Jul 9, 2026
Merged

Fix: remove some procedure that should not be included in lease frame #18167

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 1 addition & 5 deletions ...n/java/org/apache/iotdb/confignode/procedure/impl/schema/DeactivateTemplateProcedure.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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<Integer, TSStatus> statusMap = clientHandler.getResponseMap();
for (TSStatus status : statusMap.values()) {
// all dataNodes must clear the related schema cache
Expand Down
40 changes: 0 additions & 40 deletions ...nfignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/SchemaUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand All @@ -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 {
Expand Down Expand Up @@ -337,17 +335,6 @@ public static TUpdateTableReq rollbackUpdateTableReq(
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
Expand Down Expand Up @@ -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.
*
* <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;
Expand Down
64 changes: 38 additions & 26 deletions ...node/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/SetTTLProcedure.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -116,26 +115,20 @@ void setConfigNodeTTL(final ConfigNodeProcedureEnv env) {
}

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(
new MetadataException(ProcedureMessages.UPDATE_DATANODE_TTL_CACHE_FAILED)));
}
}

/**
* 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;
Expand Down Expand Up @@ -165,11 +158,7 @@ DataNodeAsyncRequestContext<TSetTTLReq, TSStatus> sendTTLRequest(
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;
}

Expand All @@ -179,6 +168,19 @@ private TSetTTLReq buildSetTTLReq(
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;
Expand Down Expand Up @@ -218,20 +220,30 @@ private void restoreTTLOnConfigNode(
}

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
Expand Down
56 changes: 26 additions & 30 deletions ...src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/SetTemplateProcedure.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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<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);
}
Expand Down Expand Up @@ -407,11 +411,7 @@ private void commitReleaseTemplate(final ConfigNodeProcedureEnv env) {
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()) {
Expand Down Expand Up @@ -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<Integer, TSStatus> statusMap = clientHandler.getResponseMap();
for (final Map.Entry<Integer, TSStatus> entry : statusMap.entrySet()) {
// all dataNodes must clear the related template cache
Expand Down
46 changes: 23 additions & 23 deletions ...c/main/java/org/apache/iotdb/confignode/procedure/impl/schema/UnsetTemplateProcedure.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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<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));
}
}
}

Expand Down Expand Up @@ -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<Integer, TSStatus> statusMap = clientHandler.getResponseMap();
for (TSStatus status : statusMap.values()) {
// all dataNodes must clear the related template cache
Expand Down
Loading
Loading