From 51c19da567adea75225e5c8079d217c3827475d0 Mon Sep 17 00:00:00 2001 From: Yaobin Chen Date: Mon, 19 Oct 2026 12:14:04 +0800 Subject: [PATCH 1/5] Fix the procedure frame that prohibiting rollback of a procedure which has passed its commit point --- .../confignode/i18n/ProcedureMessages.java | 3 + .../confignode/i18n/ProcedureMessages.java | 3 + .../iotdb/confignode/procedure/Procedure.java | 19 ++ .../procedure/ProcedureExecutor.java | 64 ++++++ .../procedure/impl/StateMachineProcedure.java | 26 +++ .../table/AbstractSetPropertiesProcedure.java | 10 + .../schema/table/AddTableColumnProcedure.java | 10 + .../AlterTableColumnDataTypeProcedure.java | 10 + .../schema/table/CreateTableProcedure.java | 10 + .../table/DropTableColumnProcedure.java | 10 + .../impl/schema/table/DropTableProcedure.java | 12 ++ .../table/RenameTableColumnProcedure.java | 10 + .../schema/table/RenameTableProcedure.java | 10 + .../schema/table/view/DropViewProcedure.java | 11 + .../impl/StateMachineProcedureTest.java | 192 ++++++++++++++++++ 15 files changed, 400 insertions(+) create mode 100644 iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/StateMachineProcedureTest.java diff --git a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java index 18aca01243908..e63805ef51c10 100644 --- a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java +++ b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java @@ -1430,4 +1430,7 @@ private ProcedureMessages() {} "DataNode setSystemStatus RPC failed: %s: %s: %s; "; public static final String EXCEPTION_CONFIGNODE_TRYSETNODESTATUS_FAILED_ARG_ARG_ARG_C9BA78F4 = "ConfigNode trySetNodeStatus failed: %s: %s: %s; "; + public static final String + LOG_PROCEDURE_ARG_IS_PAST_ITS_COMMIT_POINT_KEEPING_EVERY_STATE_IT_EXECUTED_CA21D7E9 = + "Procedure {} is past its commit point; keeping every state it executed."; } diff --git a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java index 14e52d72b9efe..43104a9ee59ed 100644 --- a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java +++ b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java @@ -1515,4 +1515,7 @@ private ProcedureMessages() {} "DataNode setSystemStatus RPC 失败：%s：%s：%s； "; public static final String EXCEPTION_CONFIGNODE_TRYSETNODESTATUS_FAILED_ARG_ARG_ARG_C9BA78F4 = "ConfigNode trySetNodeStatus 失败：%s：%s：%s； "; + public static final String + LOG_PROCEDURE_ARG_IS_PAST_ITS_COMMIT_POINT_KEEPING_EVERY_STATE_IT_EXECUTED_CA21D7E9 = + "流程 {} 已越过提交点，保留它执行过的所有状态。"; } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/Procedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/Procedure.java index 10ae561c2af57..6f8ccd7970880 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/Procedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/Procedure.java @@ -99,6 +99,25 @@ public final boolean hasLock() { protected abstract void rollback(Env env) throws IOException, InterruptedException, ProcedureException; + /** + * Whether the states this procedure has already executed may still be undone. The executor asks + * this once, before it starts to roll back the root procedure stack. A procedure which has passed + * the point after which its changes can no longer be undone returns false: none of its states is + * then rolled back, and its effects stay as they are. + */ + public boolean isRollbackAllowed() { + return true; + } + + /** + * Drops the bookkeeping of the states this procedure has executed. The persisted stack indexes + * are what a recovered rollback rebuilds its stack from, so dropping them is what makes the + * decision survive a leader switch. + */ + protected synchronized void abandonRollback() { + stackIndexes = null; + } + public void serialize(DataOutputStream stream) throws IOException { // procid stream.writeLong(this.procId); diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java index 82b217b93eab6..febe8b13b213b 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java @@ -40,8 +40,10 @@ import java.util.ArrayDeque; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.Deque; import java.util.HashSet; +import java.util.IdentityHashMap; import java.util.List; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; @@ -700,6 +702,16 @@ private LockStateResult executeRootStackRollback( () -> store.update(rootProcedure), "root procedure rollback"); } List> subprocStack = procedureStack.getSubproceduresStack(); + // Decide which procedures must keep the states they have executed before anything is rolled + // back: the decision reads the state history, which the rollback itself shrinks. Identity is + // used because procedures override equals with the data of the operation they run. + final Set> proceduresKeepingTheirStates = + Collections.newSetFromMap(new IdentityHashMap<>()); + for (Procedure procedure : subprocStack) { + if (!procedure.isSuccess() && !procedure.isRollbackAllowed()) { + proceduresKeepingTheirStates.add(procedure); + } + } int stackTail = subprocStack.size(); while (stackTail-- > 0) { Procedure procedure = subprocStack.get(stackTail); @@ -708,6 +720,10 @@ private LockStateResult executeRootStackRollback( cleanupAfterRollback(procedure); continue; } + if (proceduresKeepingTheirStates.contains(procedure)) { + // Finished below, once the rest of the stack has been rolled back. + continue; + } ProcedureLockState lockState = acquireLock(procedure); if (lockState != ProcedureLockState.LOCK_ACQUIRED) { return new LockStateResult<>(lockState, procedure); @@ -730,12 +746,60 @@ private LockStateResult executeRootStackRollback( } } + if (!proceduresKeepingTheirStates.isEmpty()) { + subprocStack.removeIf(proceduresKeepingTheirStates::contains); + proceduresKeepingTheirStates.forEach(this::finishWithoutRollback); + } + LOG.info( ProcedureMessages.ROLLED_BACK_TIME_DURATION_IS, rootProcedure, rootProcedure.elapsedTime()); rootProcedureCleanup(rootProcedure); return new LockStateResult<>(ProcedureLockState.LOCK_ACQUIRED, rootProcedure); } + /** + * Finishes a procedure without undoing any of the states it has executed. The failure is still + * reported, as the operation the procedure was part of did not complete; what is dropped is only + * the bookkeeping of the rollback, so that neither this rollback nor the one a new leader + * recovers touches this procedure again. + */ + private void finishWithoutRollback(final Procedure procedure) { + procedure.abandonRollback(); + if (!procedure.isSuccess()) { + procedure.setState(ProcedureState.ROLLEDBACK); + } + if (procedure.hasParent()) { + try { + store.delete(procedure.getProcId()); + // do not remove this procedure when exception occurred + procedures.remove(procedure.getProcId()); + } catch (final Exception e) { + LOG.warn(ProcedureMessages.FAILED_TO_DELETE_PROCEDURE_ON_ROLLBACK, e); + } + } else { + // A child which never executed has no rollback step and is not visited above, so its record + // goes with the ones of the children which were. + final long[] childProcIds = rollbackStack.get(procedure.getProcId()).getSubprocedureIds(); + if (childProcIds.length > 0) { + try { + store.delete(childProcIds); + } catch (final Exception e) { + LOG.warn(ProcedureMessages.FAILED_TO_DELETE_PROCEDURE_ON_ROLLBACK, e); + } + } + // The stack indexes and the states just dropped are persisted here, so that the rollback + // recovered by a new leader leaves this procedure out. Endless retry since this step is not + // idempotent. + RetryUtils.executeWithEndlessBackoffRetry( + () -> store.update(procedure), "finish procedure without rollback"); + } + procedure.updateMetricsOnFinish(getEnvironment(), procedure.elapsedTime(), false); + LOG.info( + ProcedureMessages + .LOG_PROCEDURE_ARG_IS_PAST_ITS_COMMIT_POINT_KEEPING_EVERY_STATE_IT_EXECUTED_CA21D7E9, + procedure); + } + private ProcedureLockState acquireLock(Procedure proc) { if (proc.hasLock()) { return ProcedureLockState.LOCK_ACQUIRED; diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/StateMachineProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/StateMachineProcedure.java index f34225cede983..9e8b46586a05b 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/StateMachineProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/StateMachineProcedure.java @@ -31,6 +31,7 @@ import java.io.IOException; import java.nio.ByteBuffer; import java.util.ArrayList; +import java.util.Iterator; import java.util.List; import java.util.concurrent.ConcurrentLinkedDeque; @@ -143,6 +144,25 @@ protected final boolean hasReachedState(final TState state) { return states.contains(getStateId(state)); } + /** + * Returns whether the specified state has been executed and completed. A state which only appears + * as the last element of the history is scheduled but has not completed yet; one which appears + * before it has completed at least once, even if it has been scheduled again since. + * + *

The commit states this is used for must not keep scheduling themselves until a condition + * holds: such a state is reported as completed as soon as it runs once. + */ + protected final boolean hasCompletedState(final TState state) { + final int stateId = getStateId(state); + final Iterator stateIds = states.iterator(); + while (stateIds.hasNext()) { + if (stateIds.next() == stateId) { + return stateIds.hasNext(); + } + } + return false; + } + /** * Add a child procedure to execute. * @@ -252,6 +272,12 @@ protected boolean isEofState() { return !states.isEmpty() && states.getLast() == EOF_STATE; } + @Override + protected synchronized void abandonRollback() { + super.abandonRollback(); + states.clear(); + } + /** * Used by the default implementation of abort() to know if the current state can be aborted and * rollback can be triggered. diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AbstractSetPropertiesProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AbstractSetPropertiesProcedure.java index be6f67007a3db..8ad2c7ec1d1b4 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AbstractSetPropertiesProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AbstractSetPropertiesProcedure.java @@ -162,6 +162,16 @@ private void setProperties(final ConfigNodeProcedureEnv env) { protected abstract ConfigPhysicalPlan createSetPropertiesPlan( Map properties, boolean isRollback); + /** + * The properties are set on the ConfigNode once SET_PROPERTIES has completed, and the ConfigNode + * schema is the source of truth the DataNodes resync from. From that point on the procedure must + * not roll back, even if a later state fails. + */ + @Override + public boolean isRollbackAllowed() { + return !hasCompletedState(SetTablePropertiesState.SET_PROPERTIES); + } + @Override protected void rollbackState( final ConfigNodeProcedureEnv env, final SetTablePropertiesState state) diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AddTableColumnProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AddTableColumnProcedure.java index 44f9e2f07f9fe..0bc4de673d42d 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AddTableColumnProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AddTableColumnProcedure.java @@ -154,6 +154,16 @@ protected String getActionMessage() { return "add table column"; } + /** + * The column is part of the table on the ConfigNode once ADD_COLUMN has completed, and the + * ConfigNode schema is the source of truth the DataNodes resync from. From that point on the + * procedure must not roll back, even if a later state fails. + */ + @Override + public boolean isRollbackAllowed() { + return !hasCompletedState(AddTableColumnState.ADD_COLUMN); + } + @Override protected void rollbackState(final ConfigNodeProcedureEnv env, final AddTableColumnState state) throws IOException, InterruptedException, ProcedureException { diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AlterTableColumnDataTypeProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AlterTableColumnDataTypeProcedure.java index 12491f24a4867..a49c4a2fc8e0b 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AlterTableColumnDataTypeProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AlterTableColumnDataTypeProcedure.java @@ -161,6 +161,16 @@ private void alterColumnDataType(final ConfigNodeProcedureEnv env) { setNextState(AlterTableColumnDataTypeState.COMMIT_RELEASE); } + /** + * The column type is changed on the ConfigNode once ALTER_TABLE_COLUMN_DATA_TYPE has completed, + * and the ConfigNode schema is the source of truth the DataNodes resync from. From that point on + * the procedure must not roll back, even if a later state fails. + */ + @Override + public boolean isRollbackAllowed() { + return !hasCompletedState(AlterTableColumnDataTypeState.ALTER_TABLE_COLUMN_DATA_TYPE); + } + @Override protected boolean isRollbackSupported(final AlterTableColumnDataTypeState state) { return false; diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/CreateTableProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/CreateTableProcedure.java index f2379ae1256eb..1a8dd99a2b257 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/CreateTableProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/CreateTableProcedure.java @@ -210,6 +210,16 @@ private void commitReleaseTable(final ConfigNodeProcedureEnv env) { } } + /** + * A table is committed on the ConfigNode once COMMIT_CREATE has completed: its status turns to + * USING, so a rollback would delete it while the DataNodes keep it. From that point on the + * procedure must not roll back, even if a later state fails. + */ + @Override + public boolean isRollbackAllowed() { + return !hasCompletedState(CreateTableState.COMMIT_CREATE); + } + @Override protected boolean isRollbackSupported(final CreateTableState state) { return true; diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableColumnProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableColumnProcedure.java index a51c636551ea7..84bea0fe251de 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableColumnProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableColumnProcedure.java @@ -233,6 +233,16 @@ private void dropColumn(final ConfigNodeProcedureEnv env) { } } + /** + * The column is removed from the ConfigNode schema once DROP_COLUMN has completed, and the + * ConfigNode schema is the source of truth the DataNodes resync from. From that point on the + * procedure must not roll back, even if a later state fails. + */ + @Override + public boolean isRollbackAllowed() { + return !hasCompletedState(DropTableColumnState.DROP_COLUMN); + } + @Override protected boolean isRollbackSupported(final DropTableColumnState state) { return false; diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableProcedure.java index cdb4d4b71c81d..a3833cd99ea9a 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableProcedure.java @@ -386,6 +386,18 @@ private void dropTable(final ConfigNodeProcedureEnv env) { } } + /** + * The data is deleted logically once DELETE_DATA has completed: the deletion is applied to every + * data region, and only the physical removal of the mod files continues in the background. A + * rollback could not bring the data back, so from that point on the procedure must not roll back, + * even if a later state fails. A view holds no data and goes straight to DROP_TABLE, which is + * checked by {@link DropViewProcedure}. + */ + @Override + public boolean isRollbackAllowed() { + return !hasCompletedState(DropTableState.DELETE_DATA); + } + @Override protected boolean isRollbackSupported(final DropTableState state) { return state == DropTableState.CHECK_AND_INVALIDATE_TABLE || state == DropTableState.PRE_DELETE; diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/RenameTableColumnProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/RenameTableColumnProcedure.java index 11d10901a352a..49bc1f8e1163c 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/RenameTableColumnProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/RenameTableColumnProcedure.java @@ -154,6 +154,16 @@ private void renameColumn(final ConfigNodeProcedureEnv env) { } } + /** + * The column is renamed on the ConfigNode once RENAME_COLUMN has completed, and the ConfigNode + * schema is the source of truth the DataNodes resync from. From that point on the procedure must + * not roll back, even if a later state fails. + */ + @Override + public boolean isRollbackAllowed() { + return !hasCompletedState(RenameTableColumnState.RENAME_COLUMN); + } + @Override protected void rollbackState(final ConfigNodeProcedureEnv env, final RenameTableColumnState state) throws IOException, InterruptedException, ProcedureException { diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/RenameTableProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/RenameTableProcedure.java index e40b041d0f6e7..553bbe3b9dcad 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/RenameTableProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/RenameTableProcedure.java @@ -144,6 +144,16 @@ private void renameTable(final ConfigNodeProcedureEnv env) { } } + /** + * The table is renamed on the ConfigNode once RENAME_TABLE has completed, and the ConfigNode + * schema is the source of truth the DataNodes resync from. From that point on the procedure must + * not roll back, even if a later state fails. + */ + @Override + public boolean isRollbackAllowed() { + return !hasCompletedState(RenameTableState.RENAME_TABLE); + } + @Override protected void rollbackState(final ConfigNodeProcedureEnv env, final RenameTableState state) throws IOException, InterruptedException, ProcedureException { diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/view/DropViewProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/view/DropViewProcedure.java index 73a7732fc87ea..dfbde4c986f35 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/view/DropViewProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/view/DropViewProcedure.java @@ -20,6 +20,7 @@ package org.apache.iotdb.confignode.procedure.impl.schema.table.view; import org.apache.iotdb.confignode.procedure.impl.schema.table.DropTableProcedure; +import org.apache.iotdb.confignode.procedure.state.schema.DropTableState; import org.apache.iotdb.confignode.procedure.store.ProcedureType; import java.io.DataOutputStream; @@ -38,6 +39,16 @@ public DropViewProcedure( super(database, tableName, queryId, isGeneratedByPipe); } + /** + * A view holds no data, so the procedure goes straight from PRE_DELETE to DROP_TABLE. The view is + * removed from the ConfigNode once DROP_TABLE has completed, and that cannot be undone, so from + * that point on the procedure must not roll back, even if a later state fails. + */ + @Override + public boolean isRollbackAllowed() { + return !hasCompletedState(DropTableState.DROP_TABLE); + } + @Override public void serialize(final DataOutputStream stream) throws IOException { stream.writeShort( diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/StateMachineProcedureTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/StateMachineProcedureTest.java new file mode 100644 index 0000000000000..b9d64d62c12fe --- /dev/null +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/StateMachineProcedureTest.java @@ -0,0 +1,192 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.confignode.procedure.impl; + +import org.apache.iotdb.confignode.procedure.entity.SimpleSTMProcedure; +import org.apache.iotdb.confignode.procedure.env.TestProcEnv; +import org.apache.iotdb.confignode.procedure.exception.ProcedureException; + +import org.junit.Assert; +import org.junit.Test; + +public class StateMachineProcedureTest { + + private enum State { + PREPARE, + APPLY, + RELEASE + } + + /** + * A procedure which must not be rolled back once {@code APPLY} has completed, the way the table + * procedures treat the state which applies their change on the ConfigNode. + */ + private static class CommitPointProcedure extends StateMachineProcedure { + + private boolean failAtApply; + + /** How many times APPLY schedules itself before it moves on. */ + private int applyRepeats; + + @Override + protected Flow executeFromState(final TestProcEnv env, final State state) { + switch (state) { + case PREPARE: + setNextState(State.APPLY); + break; + case APPLY: + if (failAtApply) { + setFailure(new ProcedureException("apply failed")); + } else if (applyRepeats-- > 0) { + // A state which keeps scheduling itself, the way one waiting for a condition does. + setNextState(State.APPLY); + } else { + setNextState(State.RELEASE); + } + break; + case RELEASE: + return Flow.NO_MORE_STATE; + default: + throw new IllegalStateException(); + } + return Flow.HAS_MORE_STATE; + } + + @Override + protected void rollbackState(final TestProcEnv env, final State state) { + // The point of the test is that this is never reached after the commit state. + } + + @Override + public boolean isRollbackAllowed() { + return !hasCompletedState(State.APPLY); + } + + @Override + protected State getState(final int stateId) { + return State.values()[stateId]; + } + + @Override + protected int getStateId(final State state) { + return state.ordinal(); + } + + @Override + protected State getInitialState() { + return State.PREPARE; + } + + /** Runs one state, the way the executor does. */ + private void runStep() throws InterruptedException { + execute(new TestProcEnv()); + } + + private boolean completed(final State state) { + return hasCompletedState(state); + } + + private State currentState() { + return getCurrentState(); + } + + private void addSlot(final int index) { + addStackIndex(index); + } + + private int[] stackIndexes() { + return getStackIndexes(); + } + + private void abandon() { + abandonRollback(); + } + } + + @Test + public void rollbackIsAllowedUntilTheCommitStateCompletes() throws InterruptedException { + final CommitPointProcedure procedure = new CommitPointProcedure(); + + // PREPARE has completed and APPLY is scheduled, but APPLY has not run yet. + procedure.runStep(); + Assert.assertTrue(procedure.completed(State.PREPARE)); + Assert.assertFalse(procedure.completed(State.APPLY)); + Assert.assertTrue(procedure.isRollbackAllowed()); + + // APPLY has completed: nothing may be undone from now on. + procedure.runStep(); + Assert.assertTrue(procedure.completed(State.APPLY)); + Assert.assertFalse(procedure.isRollbackAllowed()); + + // A later state does not change the answer. + procedure.runStep(); + Assert.assertFalse(procedure.isRollbackAllowed()); + } + + @Test + public void aCommitStateWhichRunsAgainStillCountsAsCompleted() throws InterruptedException { + final CommitPointProcedure procedure = new CommitPointProcedure(); + procedure.applyRepeats = 1; + procedure.runStep(); // PREPARE + procedure.runStep(); // APPLY, which schedules APPLY again + + // The last element is the APPLY which has not run yet, but the one before it has completed. + Assert.assertEquals(State.APPLY, procedure.currentState()); + Assert.assertTrue(procedure.completed(State.APPLY)); + Assert.assertFalse(procedure.isRollbackAllowed()); + } + + @Test + public void aCommitStateWhichFailsDoesNotBlockTheRollback() throws InterruptedException { + final CommitPointProcedure procedure = new CommitPointProcedure(); + procedure.runStep(); + + procedure.failAtApply = true; + procedure.runStep(); + + Assert.assertFalse(procedure.completed(State.APPLY)); + Assert.assertTrue(procedure.isRollbackAllowed()); + } + + @Test + public void abandoningTheRollbackDropsTheStackSlotsAndTheStateHistory() + throws InterruptedException { + final CommitPointProcedure procedure = new CommitPointProcedure(); + procedure.runStep(); + procedure.runStep(); + procedure.addSlot(0); + procedure.addSlot(1); + procedure.addSlot(2); + Assert.assertEquals(3, procedure.stackIndexes().length); + Assert.assertFalse(procedure.isRollbackAllowed()); + + procedure.abandon(); + + // A recovered rollback rebuilds its stack from the persisted indexes, so they must be gone. + Assert.assertNull(procedure.stackIndexes()); + // An empty history reports the initial state, which is how a drained history shows up. + Assert.assertEquals(State.PREPARE, procedure.currentState()); + } + + @Test + public void aProcedureWithoutACommitStateIsAlwaysRollbackable() { + Assert.assertTrue(new SimpleSTMProcedure().isRollbackAllowed()); + } +} From 754625cce269130d339cc56fff800f90a14408f4 Mon Sep 17 00:00:00 2001 From: Yaobin Chen Date: Mon, 19 Oct 2026 14:40:00 +0800 Subject: [PATCH 2/5] Remove the abundant api --- .../iotdb/confignode/procedure/Procedure.java | 7 +- .../procedure/ProcedureExecutor.java | 2 +- .../procedure/impl/StateMachineProcedure.java | 56 +++++++++------ .../impl/node/RemoveAINodeProcedure.java | 9 +++ .../table/AbstractSetPropertiesProcedure.java | 15 ++-- .../schema/table/AddTableColumnProcedure.java | 15 ++-- .../AlterTableColumnDataTypeProcedure.java | 11 +-- .../schema/table/CreateTableProcedure.java | 20 +++--- .../table/DropTableColumnProcedure.java | 11 +-- .../impl/schema/table/DropTableProcedure.java | 21 +++--- .../table/RenameTableColumnProcedure.java | 15 ++-- .../schema/table/RenameTableProcedure.java | 15 ++-- .../schema/table/view/DropViewProcedure.java | 15 ++-- .../impl/StateMachineProcedureTest.java | 71 +++++++++---------- 14 files changed, 156 insertions(+), 127 deletions(-) diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/Procedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/Procedure.java index 6f8ccd7970880..dc26ced3b6462 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/Procedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/Procedure.java @@ -101,11 +101,10 @@ protected abstract void rollback(Env env) /** * Whether the states this procedure has already executed may still be undone. The executor asks - * this once, before it starts to roll back the root procedure stack. A procedure which has passed - * the point after which its changes can no longer be undone returns false: none of its states is - * then rolled back, and its effects stay as they are. + * this once, before it starts to roll back the root procedure stack. A procedure which returns + * false is finished without undoing any of its states, see {@link #abandonRollback()}. */ - public boolean isRollbackAllowed() { + protected boolean isRollbackSupported() { return true; } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java index febe8b13b213b..abe40df847cf7 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java @@ -708,7 +708,7 @@ private LockStateResult executeRootStackRollback( final Set> proceduresKeepingTheirStates = Collections.newSetFromMap(new IdentityHashMap<>()); for (Procedure procedure : subprocStack) { - if (!procedure.isSuccess() && !procedure.isRollbackAllowed()) { + if (!procedure.isSuccess() && !procedure.isRollbackSupported()) { proceduresKeepingTheirStates.add(procedure); } } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/StateMachineProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/StateMachineProcedure.java index 9e8b46586a05b..5f518f728e936 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/StateMachineProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/StateMachineProcedure.java @@ -144,25 +144,6 @@ protected final boolean hasReachedState(final TState state) { return states.contains(getStateId(state)); } - /** - * Returns whether the specified state has been executed and completed. A state which only appears - * as the last element of the history is scheduled but has not completed yet; one which appears - * before it has completed at least once, even if it has been scheduled again since. - * - *

The commit states this is used for must not keep scheduling themselves until a condition - * holds: such a state is reported as completed as soon as it runs once. - */ - protected final boolean hasCompletedState(final TState state) { - final int stateId = getStateId(state); - final Iterator stateIds = states.iterator(); - while (stateIds.hasNext()) { - if (stateIds.next() == stateId) { - return stateIds.hasNext(); - } - } - return false; - } - /** * Add a child procedure to execute. * @@ -279,11 +260,42 @@ protected synchronized void abandonRollback() { } /** - * Used by the default implementation of abort() to know if the current state can be aborted and - * rollback can be triggered. + * Whether a failure in the specified state can be rolled back. A procedure asks this in its error + * handling to tell a fatal failure from a retriable one; the executor asks it once, through the + * overload without a state, to decide whether the procedure may be rolled back at all. + * + *

The executor asks with the state the procedure is at: the state which failed or is + * scheduled, or the last state executed when every state has already run. A procedure which + * returns false for the states after its commit point is therefore not rolled back once that + * point has been passed. */ protected boolean isRollbackSupported(final TState state) { - return false; + return true; + } + + /** + * Whether the states this procedure has executed may still be rolled back. The answer comes from + * the state the procedure is at, through the per-state overload above. + */ + @Override + protected final boolean isRollbackSupported() { + return isRollbackSupported(getRollbackDecisionState()); + } + + /** + * The state a rollback decision is taken at. The last element of the history is the state which + * failed or is scheduled; once every state has run, the end marker is last and the decision falls + * on the state before it. + */ + private TState getRollbackDecisionState() { + final Iterator stateIds = states.descendingIterator(); + while (stateIds.hasNext()) { + final int stateId = stateIds.next(); + if (stateId != EOF_STATE) { + return getState(stateId); + } + } + return getInitialState(); } private boolean noMoreState() { diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/node/RemoveAINodeProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/node/RemoveAINodeProcedure.java index 23b041c2587e9..a6fdd7d61517e 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/node/RemoveAINodeProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/node/RemoveAINodeProcedure.java @@ -135,6 +135,15 @@ protected void rollbackState( // no need to rollback } + /** + * A failure in either state is retried instead of failing the procedure, which has nothing to + * roll back. + */ + @Override + protected boolean isRollbackSupported(final RemoveAINodeState state) { + return false; + } + @Override protected RemoveAINodeState getState(int stateId) { return RemoveAINodeState.values()[stateId]; diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AbstractSetPropertiesProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AbstractSetPropertiesProcedure.java index 8ad2c7ec1d1b4..a354f0ff3a393 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AbstractSetPropertiesProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AbstractSetPropertiesProcedure.java @@ -164,12 +164,19 @@ protected abstract ConfigPhysicalPlan createSetPropertiesPlan( /** * The properties are set on the ConfigNode once SET_PROPERTIES has completed, and the ConfigNode - * schema is the source of truth the DataNodes resync from. From that point on the procedure must - * not roll back, even if a later state fails. + * schema is the source of truth the DataNodes resync from. The procedure therefore stays + * rollbackable up to SET_PROPERTIES and not after it. */ @Override - public boolean isRollbackAllowed() { - return !hasCompletedState(SetTablePropertiesState.SET_PROPERTIES); + protected boolean isRollbackSupported(final SetTablePropertiesState state) { + switch (state) { + case VALIDATE_TABLE: + case PRE_RELEASE: + case SET_PROPERTIES: + return true; + default: + return false; + } } @Override diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AddTableColumnProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AddTableColumnProcedure.java index 0bc4de673d42d..07bdbf44c55c9 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AddTableColumnProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AddTableColumnProcedure.java @@ -156,12 +156,19 @@ protected String getActionMessage() { /** * The column is part of the table on the ConfigNode once ADD_COLUMN has completed, and the - * ConfigNode schema is the source of truth the DataNodes resync from. From that point on the - * procedure must not roll back, even if a later state fails. + * ConfigNode schema is the source of truth the DataNodes resync from. The procedure therefore + * stays rollbackable up to ADD_COLUMN and not after it. */ @Override - public boolean isRollbackAllowed() { - return !hasCompletedState(AddTableColumnState.ADD_COLUMN); + protected boolean isRollbackSupported(final AddTableColumnState state) { + switch (state) { + case COLUMN_CHECK: + case PRE_RELEASE: + case ADD_COLUMN: + return true; + default: + return false; + } } @Override diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AlterTableColumnDataTypeProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AlterTableColumnDataTypeProcedure.java index a49c4a2fc8e0b..1cfda63fcf736 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AlterTableColumnDataTypeProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AlterTableColumnDataTypeProcedure.java @@ -161,16 +161,7 @@ private void alterColumnDataType(final ConfigNodeProcedureEnv env) { setNextState(AlterTableColumnDataTypeState.COMMIT_RELEASE); } - /** - * The column type is changed on the ConfigNode once ALTER_TABLE_COLUMN_DATA_TYPE has completed, - * and the ConfigNode schema is the source of truth the DataNodes resync from. From that point on - * the procedure must not roll back, even if a later state fails. - */ - @Override - public boolean isRollbackAllowed() { - return !hasCompletedState(AlterTableColumnDataTypeState.ALTER_TABLE_COLUMN_DATA_TYPE); - } - + /** The rollback of this procedure is a no-op, so it is never rolled back. */ @Override protected boolean isRollbackSupported(final AlterTableColumnDataTypeState state) { return false; diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/CreateTableProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/CreateTableProcedure.java index 1a8dd99a2b257..ac573463ebb23 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/CreateTableProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/CreateTableProcedure.java @@ -210,19 +210,17 @@ private void commitReleaseTable(final ConfigNodeProcedureEnv env) { } } - /** - * A table is committed on the ConfigNode once COMMIT_CREATE has completed: its status turns to - * USING, so a rollback would delete it while the DataNodes keep it. From that point on the - * procedure must not roll back, even if a later state fails. - */ - @Override - public boolean isRollbackAllowed() { - return !hasCompletedState(CreateTableState.COMMIT_CREATE); - } - @Override protected boolean isRollbackSupported(final CreateTableState state) { - return true; + switch (state) { + case CHECK_TABLE_EXISTENCE: + case PRE_CREATE: + case PRE_RELEASE: + case COMMIT_CREATE: + return true; + default: + return false; + } } @Override diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableColumnProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableColumnProcedure.java index 84bea0fe251de..75000e36ef67a 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableColumnProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableColumnProcedure.java @@ -233,16 +233,7 @@ private void dropColumn(final ConfigNodeProcedureEnv env) { } } - /** - * The column is removed from the ConfigNode schema once DROP_COLUMN has completed, and the - * ConfigNode schema is the source of truth the DataNodes resync from. From that point on the - * procedure must not roll back, even if a later state fails. - */ - @Override - public boolean isRollbackAllowed() { - return !hasCompletedState(DropTableColumnState.DROP_COLUMN); - } - + /** The rollback of this procedure is a no-op, so it is never rolled back. */ @Override protected boolean isRollbackSupported(final DropTableColumnState state) { return false; diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableProcedure.java index a3833cd99ea9a..54217212a1151 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableProcedure.java @@ -388,19 +388,20 @@ private void dropTable(final ConfigNodeProcedureEnv env) { /** * The data is deleted logically once DELETE_DATA has completed: the deletion is applied to every - * data region, and only the physical removal of the mod files continues in the background. A - * rollback could not bring the data back, so from that point on the procedure must not roll back, - * even if a later state fails. A view holds no data and goes straight to DROP_TABLE, which is - * checked by {@link DropViewProcedure}. + * data region, and only the physical removal of the mod files continues in the background. The + * procedure is therefore not rolled back once it has moved on to DELETE_DEVICES. A view holds no + * data and goes straight to DROP_TABLE, see {@link DropViewProcedure}. */ - @Override - public boolean isRollbackAllowed() { - return !hasCompletedState(DropTableState.DELETE_DATA); - } - @Override protected boolean isRollbackSupported(final DropTableState state) { - return state == DropTableState.CHECK_AND_INVALIDATE_TABLE || state == DropTableState.PRE_DELETE; + switch (state) { + case CHECK_AND_INVALIDATE_TABLE: + case PRE_DELETE: + case DELETE_DATA: + return true; + default: + return false; + } } @Override diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/RenameTableColumnProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/RenameTableColumnProcedure.java index 49bc1f8e1163c..03088935ce616 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/RenameTableColumnProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/RenameTableColumnProcedure.java @@ -156,12 +156,19 @@ private void renameColumn(final ConfigNodeProcedureEnv env) { /** * The column is renamed on the ConfigNode once RENAME_COLUMN has completed, and the ConfigNode - * schema is the source of truth the DataNodes resync from. From that point on the procedure must - * not roll back, even if a later state fails. + * schema is the source of truth the DataNodes resync from. The procedure therefore stays + * rollbackable up to RENAME_COLUMN and not after it. */ @Override - public boolean isRollbackAllowed() { - return !hasCompletedState(RenameTableColumnState.RENAME_COLUMN); + protected boolean isRollbackSupported(final RenameTableColumnState state) { + switch (state) { + case COLUMN_CHECK: + case PRE_RELEASE: + case RENAME_COLUMN: + return true; + default: + return false; + } } @Override diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/RenameTableProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/RenameTableProcedure.java index 553bbe3b9dcad..5296253304913 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/RenameTableProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/RenameTableProcedure.java @@ -146,12 +146,19 @@ private void renameTable(final ConfigNodeProcedureEnv env) { /** * The table is renamed on the ConfigNode once RENAME_TABLE has completed, and the ConfigNode - * schema is the source of truth the DataNodes resync from. From that point on the procedure must - * not roll back, even if a later state fails. + * schema is the source of truth the DataNodes resync from. The procedure therefore stays + * rollbackable up to RENAME_TABLE and not after it. */ @Override - public boolean isRollbackAllowed() { - return !hasCompletedState(RenameTableState.RENAME_TABLE); + protected boolean isRollbackSupported(final RenameTableState state) { + switch (state) { + case COLUMN_CHECK: + case PRE_RELEASE: + case RENAME_TABLE: + return true; + default: + return false; + } } @Override diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/view/DropViewProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/view/DropViewProcedure.java index dfbde4c986f35..163150c0a09e4 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/view/DropViewProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/view/DropViewProcedure.java @@ -41,12 +41,19 @@ public DropViewProcedure( /** * A view holds no data, so the procedure goes straight from PRE_DELETE to DROP_TABLE. The view is - * removed from the ConfigNode once DROP_TABLE has completed, and that cannot be undone, so from - * that point on the procedure must not roll back, even if a later state fails. + * removed from the ConfigNode once DROP_TABLE has completed; before that it can still be + * restored, so the procedure is not rolled back from COMMIT_DELETE on. */ @Override - public boolean isRollbackAllowed() { - return !hasCompletedState(DropTableState.DROP_TABLE); + protected boolean isRollbackSupported(final DropTableState state) { + switch (state) { + case CHECK_AND_INVALIDATE_TABLE: + case PRE_DELETE: + case DROP_TABLE: + return true; + default: + return false; + } } @Override diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/StateMachineProcedureTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/StateMachineProcedureTest.java index b9d64d62c12fe..340f1b88c5cda 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/StateMachineProcedureTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/StateMachineProcedureTest.java @@ -35,15 +35,13 @@ private enum State { } /** - * A procedure which must not be rolled back once {@code APPLY} has completed, the way the table - * procedures treat the state which applies their change on the ConfigNode. + * A procedure whose commit point is the completion of {@code APPLY}: it stays rollbackable while + * it is at {@code PREPARE} or {@code APPLY}, and not once it has moved on to {@code RELEASE}. */ private static class CommitPointProcedure extends StateMachineProcedure { private boolean failAtApply; - - /** How many times APPLY schedules itself before it moves on. */ - private int applyRepeats; + private boolean failAtRelease; @Override protected Flow executeFromState(final TestProcEnv env, final State state) { @@ -54,14 +52,14 @@ protected Flow executeFromState(final TestProcEnv env, final State state) { case APPLY: if (failAtApply) { setFailure(new ProcedureException("apply failed")); - } else if (applyRepeats-- > 0) { - // A state which keeps scheduling itself, the way one waiting for a condition does. - setNextState(State.APPLY); } else { setNextState(State.RELEASE); } break; case RELEASE: + if (failAtRelease) { + setFailure(new ProcedureException("release failed")); + } return Flow.NO_MORE_STATE; default: throw new IllegalStateException(); @@ -71,12 +69,12 @@ protected Flow executeFromState(final TestProcEnv env, final State state) { @Override protected void rollbackState(final TestProcEnv env, final State state) { - // The point of the test is that this is never reached after the commit state. + // The point of the test is that this is never called once the commit point has been passed. } @Override - public boolean isRollbackAllowed() { - return !hasCompletedState(State.APPLY); + protected boolean isRollbackSupported(final State state) { + return state != State.RELEASE; } @Override @@ -99,10 +97,6 @@ private void runStep() throws InterruptedException { execute(new TestProcEnv()); } - private boolean completed(final State state) { - return hasCompletedState(state); - } - private State currentState() { return getCurrentState(); } @@ -121,48 +115,47 @@ private void abandon() { } @Test - public void rollbackIsAllowedUntilTheCommitStateCompletes() throws InterruptedException { + public void rollbackIsAllowedUntilTheStateAfterTheCommitPointIsReached() + throws InterruptedException { final CommitPointProcedure procedure = new CommitPointProcedure(); - // PREPARE has completed and APPLY is scheduled, but APPLY has not run yet. + // PREPARE has completed and APPLY is scheduled. procedure.runStep(); - Assert.assertTrue(procedure.completed(State.PREPARE)); - Assert.assertFalse(procedure.completed(State.APPLY)); - Assert.assertTrue(procedure.isRollbackAllowed()); + Assert.assertEquals(State.APPLY, procedure.currentState()); + Assert.assertTrue(procedure.isRollbackSupported()); - // APPLY has completed: nothing may be undone from now on. + // APPLY has completed: the procedure is at RELEASE and must not be rolled back any more. procedure.runStep(); - Assert.assertTrue(procedure.completed(State.APPLY)); - Assert.assertFalse(procedure.isRollbackAllowed()); + Assert.assertEquals(State.RELEASE, procedure.currentState()); + Assert.assertFalse(procedure.isRollbackSupported()); - // A later state does not change the answer. + // Every state has run; the decision falls on the last state which ran, not on the end marker. procedure.runStep(); - Assert.assertFalse(procedure.isRollbackAllowed()); + Assert.assertFalse(procedure.isRollbackSupported()); } @Test - public void aCommitStateWhichRunsAgainStillCountsAsCompleted() throws InterruptedException { + public void aFailureBeforeTheCommitPointStillAllowsTheRollback() throws InterruptedException { final CommitPointProcedure procedure = new CommitPointProcedure(); - procedure.applyRepeats = 1; - procedure.runStep(); // PREPARE - procedure.runStep(); // APPLY, which schedules APPLY again + procedure.runStep(); + + procedure.failAtApply = true; + procedure.runStep(); - // The last element is the APPLY which has not run yet, but the one before it has completed. Assert.assertEquals(State.APPLY, procedure.currentState()); - Assert.assertTrue(procedure.completed(State.APPLY)); - Assert.assertFalse(procedure.isRollbackAllowed()); + Assert.assertTrue(procedure.isRollbackSupported()); } @Test - public void aCommitStateWhichFailsDoesNotBlockTheRollback() throws InterruptedException { + public void aFailureAfterTheCommitPointDoesNotAllowTheRollback() throws InterruptedException { final CommitPointProcedure procedure = new CommitPointProcedure(); procedure.runStep(); + procedure.runStep(); - procedure.failAtApply = true; + procedure.failAtRelease = true; procedure.runStep(); - Assert.assertFalse(procedure.completed(State.APPLY)); - Assert.assertTrue(procedure.isRollbackAllowed()); + Assert.assertFalse(procedure.isRollbackSupported()); } @Test @@ -175,7 +168,7 @@ public void abandoningTheRollbackDropsTheStackSlotsAndTheStateHistory() procedure.addSlot(1); procedure.addSlot(2); Assert.assertEquals(3, procedure.stackIndexes().length); - Assert.assertFalse(procedure.isRollbackAllowed()); + Assert.assertFalse(procedure.isRollbackSupported()); procedure.abandon(); @@ -186,7 +179,7 @@ public void abandoningTheRollbackDropsTheStackSlotsAndTheStateHistory() } @Test - public void aProcedureWithoutACommitStateIsAlwaysRollbackable() { - Assert.assertTrue(new SimpleSTMProcedure().isRollbackAllowed()); + public void aProcedureWhichOverridesNothingIsRollbackableEverywhere() { + Assert.assertTrue(new SimpleSTMProcedure().isRollbackSupported()); } } From eed5db94367f229856e70f5f2a582583a1815a55 Mon Sep 17 00:00:00 2001 From: Yaobin Chen Date: Mon, 19 Oct 2026 17:13:30 +0800 Subject: [PATCH 3/5] fixed the lock release --- .../iotdb/confignode/i18n/ProcedureMessages.java | 2 ++ .../iotdb/confignode/i18n/ProcedureMessages.java | 1 + .../confignode/procedure/ProcedureExecutor.java | 12 +++++++++++- 3 files changed, 14 insertions(+), 1 deletion(-) diff --git a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java index e63805ef51c10..b86093383c197 100644 --- a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java +++ b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java @@ -1433,4 +1433,6 @@ private ProcedureMessages() {} public static final String LOG_PROCEDURE_ARG_IS_PAST_ITS_COMMIT_POINT_KEEPING_EVERY_STATE_IT_EXECUTED_CA21D7E9 = "Procedure {} is past its commit point; keeping every state it executed."; + public static final String LOG_FINISH_PROCEDURE_WITHOUT_ROLLBACK_41F7A354 = + "finish procedure without rollback"; } diff --git a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java index 43104a9ee59ed..0147a4f497693 100644 --- a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java +++ b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java @@ -1518,4 +1518,5 @@ private ProcedureMessages() {} public static final String LOG_PROCEDURE_ARG_IS_PAST_ITS_COMMIT_POINT_KEEPING_EVERY_STATE_IT_EXECUTED_CA21D7E9 = "流程 {} 已越过提交点，保留它执行过的所有状态。"; + public static final String LOG_FINISH_PROCEDURE_WITHOUT_ROLLBACK_41F7A354 = "结束流程（不回滚）"; } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java index abe40df847cf7..9e7a5372a8c7e 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java @@ -768,6 +768,15 @@ private void finishWithoutRollback(final Procedure procedure) { if (!procedure.isSuccess()) { procedure.setState(ProcedureState.ROLLEDBACK); } + // The rollback loop skipped acquireLock() for this procedure, so a lock restored by + // restoreLock() on load is still recorded in lockedWhenLoading instead of locked. Finish that + // bookkeeping and release the lock here: rootProcedureCleanup() releases only the locks that + // hasLock() reports, and a finished procedure must not keep holding its lock. + if (procedure.hasLock() || procedure.isLockedWhenLoading()) { + if (acquireLock(procedure) == ProcedureLockState.LOCK_ACQUIRED) { + releaseLock(procedure, false); + } + } if (procedure.hasParent()) { try { store.delete(procedure.getProcId()); @@ -791,7 +800,8 @@ private void finishWithoutRollback(final Procedure procedure) { // recovered by a new leader leaves this procedure out. Endless retry since this step is not // idempotent. RetryUtils.executeWithEndlessBackoffRetry( - () -> store.update(procedure), "finish procedure without rollback"); + () -> store.update(procedure), + ProcedureMessages.LOG_FINISH_PROCEDURE_WITHOUT_ROLLBACK_41F7A354); } procedure.updateMetricsOnFinish(getEnvironment(), procedure.elapsedTime(), false); LOG.info( From 7fd9f8bd8301bf3b338d78851f6e5d52b16c364f Mon Sep 17 00:00:00 2001 From: Yaobin Chen Date: Tue, 20 Oct 2026 12:24:52 +0800 Subject: [PATCH 4/5] fix01 --- .../confignode/i18n/ProcedureMessages.java | 2 + .../confignode/i18n/ProcedureMessages.java | 2 + .../procedure/ProcedureExecutor.java | 78 ++++++++++++++----- .../procedure/RootProcedureStack.java | 4 +- 4 files changed, 64 insertions(+), 22 deletions(-) diff --git a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java index b86093383c197..8845eeebaa794 100644 --- a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java +++ b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java @@ -1435,4 +1435,6 @@ private ProcedureMessages() {} "Procedure {} is past its commit point; keeping every state it executed."; public static final String LOG_FINISH_PROCEDURE_WITHOUT_ROLLBACK_41F7A354 = "finish procedure without rollback"; + public static final String LOG_DELETE_CHILD_PROCEDURES_WITHOUT_ROLLBACK_6E62D1C1 = + "delete child procedures without rollback"; } diff --git a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java index 0147a4f497693..76a88916a180d 100644 --- a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java +++ b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java @@ -1519,4 +1519,6 @@ private ProcedureMessages() {} LOG_PROCEDURE_ARG_IS_PAST_ITS_COMMIT_POINT_KEEPING_EVERY_STATE_IT_EXECUTED_CA21D7E9 = "流程 {} 已越过提交点，保留它执行过的所有状态。"; public static final String LOG_FINISH_PROCEDURE_WITHOUT_ROLLBACK_41F7A354 = "结束流程（不回滚）"; + public static final String LOG_DELETE_CHILD_PROCEDURES_WITHOUT_ROLLBACK_6E62D1C1 = + "删除子流程（不回滚）"; } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java index 9e7a5372a8c7e..3408b1c750c76 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java @@ -708,13 +708,18 @@ private LockStateResult executeRootStackRollback( final Set> proceduresKeepingTheirStates = Collections.newSetFromMap(new IdentityHashMap<>()); for (Procedure procedure : subprocStack) { - if (!procedure.isSuccess() && !procedure.isRollbackSupported()) { + // A null slot is a procedure whose record was deleted before the snapshot this stack was + // rebuilt from, so there is nothing left to roll back for it. + if (procedure != null && !procedure.isSuccess() && !procedure.isRollbackSupported()) { proceduresKeepingTheirStates.add(procedure); } } int stackTail = subprocStack.size(); while (stackTail-- > 0) { Procedure procedure = subprocStack.get(stackTail); + if (procedure == null) { + continue; + } if (procedure.isSuccess()) { subprocStack.remove(stackTail); cleanupAfterRollback(procedure); @@ -748,7 +753,19 @@ private LockStateResult executeRootStackRollback( if (!proceduresKeepingTheirStates.isEmpty()) { subprocStack.removeIf(proceduresKeepingTheirStates::contains); - proceduresKeepingTheirStates.forEach(this::finishWithoutRollback); + // The children go first: releasing a child's lock writes its record back, which has to be + // covered by the root's cleanup. The root is finished last so that no write of a child + // follows the write which persists the root as finished. + for (Procedure procedure : proceduresKeepingTheirStates) { + if (procedure.hasParent()) { + finishWithoutRollback(procedure); + } + } + for (Procedure procedure : proceduresKeepingTheirStates) { + if (!procedure.hasParent()) { + finishWithoutRollback(procedure); + } + } } LOG.info( @@ -764,20 +781,19 @@ private LockStateResult executeRootStackRollback( * recovers touches this procedure again. */ private void finishWithoutRollback(final Procedure procedure) { - procedure.abandonRollback(); - if (!procedure.isSuccess()) { - procedure.setState(ProcedureState.ROLLEDBACK); - } - // The rollback loop skipped acquireLock() for this procedure, so a lock restored by - // restoreLock() on load is still recorded in lockedWhenLoading instead of locked. Finish that - // bookkeeping and release the lock here: rootProcedureCleanup() releases only the locks that - // hasLock() reports, and a finished procedure must not keep holding its lock. - if (procedure.hasLock() || procedure.isLockedWhenLoading()) { - if (acquireLock(procedure) == ProcedureLockState.LOCK_ACQUIRED) { - releaseLock(procedure, false); - } - } + // The unlock persists the procedure through doReleaseLock(). It is done first, while the + // procedure is neither marked finished nor stripped of its state history, so that a snapshot + // taken right after this write is still a state recovery resumes. Forcing the release matters + // because the procedure is not finished yet, so releaseLock() would otherwise keep the lock of + // procedures which hold it across steps. + releaseLockIfHeld(procedure); if (procedure.hasParent()) { + // Mark the procedure finished before dropping its state history, so that a worker which + // picks it up concurrently returns at the isFinished() check instead of rolling it back. + if (!procedure.isSuccess()) { + procedure.setState(ProcedureState.ROLLEDBACK); + } + procedure.abandonRollback(); try { store.delete(procedure.getProcId()); // do not remove this procedure when exception occurred @@ -788,14 +804,19 @@ private void finishWithoutRollback(final Procedure procedure) { } else { // A child which never executed has no rollback step and is not visited above, so its record // goes with the ones of the children which were. + // Every child record has to be gone before the root may be persisted as finished: a snapshot + // holding a finished root beside a child record makes recover() fail on that child. Deleting + // is idempotent, so it is retried until durable. final long[] childProcIds = rollbackStack.get(procedure.getProcId()).getSubprocedureIds(); - if (childProcIds.length > 0) { - try { - store.delete(childProcIds); - } catch (final Exception e) { - LOG.warn(ProcedureMessages.FAILED_TO_DELETE_PROCEDURE_ON_ROLLBACK, e); - } + for (final long childProcId : childProcIds) { + RetryUtils.executeWithEndlessBackoffRetry( + () -> store.delete(childProcId), + ProcedureMessages.LOG_DELETE_CHILD_PROCEDURES_WITHOUT_ROLLBACK_6E62D1C1); } + if (!procedure.isSuccess()) { + procedure.setState(ProcedureState.ROLLEDBACK); + } + procedure.abandonRollback(); // The stack indexes and the states just dropped are persisted here, so that the rollback // recovered by a new leader leaves this procedure out. Endless retry since this step is not // idempotent. @@ -810,6 +831,21 @@ private void finishWithoutRollback(final Procedure procedure) { procedure); } + /** + * The rollback loop skipped acquireLock() for procedures finished without rollback, so a lock + * restored by restoreLock() on load is still recorded in lockedWhenLoading instead of locked. + * Finish that bookkeeping and release the lock here: rootProcedureCleanup() releases only the + * locks that hasLock() reports, and a procedure finished without rollback must not keep holding + * its lock. + */ + private void releaseLockIfHeld(final Procedure procedure) { + if (procedure.hasLock() || procedure.isLockedWhenLoading()) { + if (acquireLock(procedure) == ProcedureLockState.LOCK_ACQUIRED) { + releaseLock(procedure, true); + } + } + } + private ProcedureLockState acquireLock(Procedure proc) { if (proc.hasLock()) { return ProcedureLockState.LOCK_ACQUIRED; diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/RootProcedureStack.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/RootProcedureStack.java index 1fa240997f764..db38c6175bb08 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/RootProcedureStack.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/RootProcedureStack.java @@ -89,7 +89,9 @@ protected synchronized List> getSubproceduresStack() { protected synchronized ProcedureException getException() { if (subprocStack != null) { for (Procedure proc : subprocStack) { - if (proc.hasException()) { + // Null slots come from procedures whose records were deleted before the stack was + // rebuilt, see ProcedureExecutor#executeRootStackRollback. + if (proc != null && proc.hasException()) { return proc.getException(); } } From 9fa62fc8d182a3b55da600e442b2060589cb2ba9 Mon Sep 17 00:00:00 2001 From: Yaobin Chen Date: Tue, 20 Oct 2026 15:36:16 +0800 Subject: [PATCH 5/5] wip02 --- .../confignode/procedure/ProcedureExecutor.java | 16 ++++++++++++---- 1 file changed, 12 insertions(+), 4 deletions(-) diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java index 3408b1c750c76..0b5460486b29c 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java @@ -372,7 +372,7 @@ public boolean removeInternalProcedure(InternalProcedure internalProcedure) { * * @param proc procedure */ - private void executeProcedure(Procedure proc) { + private void executeProcedure(Procedure proc) throws InterruptedException { if (proc.isFinished()) { LOG.debug(ProcedureMessages.IS_ALREADY_FINISHED, proc); return; @@ -691,7 +691,7 @@ private void yieldProcedure(Procedure proc) { * @return lock state */ private LockStateResult executeRootStackRollback( - Long rootProcId, RootProcedureStack procedureStack) { + Long rootProcId, RootProcedureStack procedureStack) throws InterruptedException { Procedure rootProcedure = procedures.get(rootProcId); ProcedureException exception = rootProcedure.getException(); if (exception == null) { @@ -752,7 +752,6 @@ private LockStateResult executeRootStackRollback( } if (!proceduresKeepingTheirStates.isEmpty()) { - subprocStack.removeIf(proceduresKeepingTheirStates::contains); // The children go first: releasing a child's lock writes its record back, which has to be // covered by the root's cleanup. The root is finished last so that no write of a child // follows the write which persists the root as finished. @@ -766,6 +765,7 @@ private LockStateResult executeRootStackRollback( finishWithoutRollback(procedure); } } + subprocStack.removeIf(proceduresKeepingTheirStates::contains); } LOG.info( @@ -780,7 +780,7 @@ private LockStateResult executeRootStackRollback( * the bookkeeping of the rollback, so that neither this rollback nor the one a new leader * recovers touches this procedure again. */ - private void finishWithoutRollback(final Procedure procedure) { + private void finishWithoutRollback(final Procedure procedure) throws InterruptedException { // The unlock persists the procedure through doReleaseLock(). It is done first, while the // procedure is neither marked finished nor stripped of its state history, so that a snapshot // taken right after this write is still a state recovery resumes. Forcing the release matters @@ -812,6 +812,11 @@ private void finishWithoutRollback(final Procedure procedure) { RetryUtils.executeWithEndlessBackoffRetry( () -> store.delete(childProcId), ProcedureMessages.LOG_DELETE_CHILD_PROCEDURES_WITHOUT_ROLLBACK_6E62D1C1); + // The retry helper also returns on interruption. Abort the entire rollback before + // marking the root finished or letting the caller clean it up. + if (Thread.currentThread().isInterrupted()) { + throw new InterruptedException(); + } } if (!procedure.isSuccess()) { procedure.setState(ProcedureState.ROLLEDBACK); @@ -823,6 +828,9 @@ private void finishWithoutRollback(final Procedure procedure) { RetryUtils.executeWithEndlessBackoffRetry( () -> store.update(procedure), ProcedureMessages.LOG_FINISH_PROCEDURE_WITHOUT_ROLLBACK_41F7A354); + if (Thread.currentThread().isInterrupted()) { + throw new InterruptedException(); + } } procedure.updateMetricsOnFinish(getEnvironment(), procedure.elapsedTime(), false); LOG.info(