This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new c75e9a8f70b Prohibit rollback of a procedure which has passed its
commit point (#18909)
c75e9a8f70b is described below
commit c75e9a8f70bcad4148a993c4d89dedd803da857b
Author: alpass163 <[email protected]>
AuthorDate: Tue Oct 20 16:41:34 2026 +0800
Prohibit rollback of a procedure which has passed its commit point (#18909)
---
.../iotdb/confignode/i18n/ProcedureMessages.java | 7 +
.../iotdb/confignode/i18n/ProcedureMessages.java | 6 +
.../iotdb/confignode/procedure/Procedure.java | 18 ++
.../confignode/procedure/ProcedureExecutor.java | 122 +++++++++++++-
.../confignode/procedure/RootProcedureStack.java | 4 +-
.../procedure/impl/StateMachineProcedure.java | 44 ++++-
.../procedure/impl/node/RemoveAINodeProcedure.java | 9 +
.../table/AbstractSetPropertiesProcedure.java | 17 ++
.../impl/schema/table/AddTableColumnProcedure.java | 17 ++
.../table/AlterTableColumnDataTypeProcedure.java | 1 +
.../impl/schema/table/CreateTableProcedure.java | 10 +-
.../schema/table/DropTableColumnProcedure.java | 1 +
.../impl/schema/table/DropTableProcedure.java | 15 +-
.../schema/table/RenameTableColumnProcedure.java | 17 ++
.../impl/schema/table/RenameTableProcedure.java | 17 ++
.../impl/schema/table/view/DropViewProcedure.java | 18 ++
.../procedure/impl/StateMachineProcedureTest.java | 185 +++++++++++++++++++++
17 files changed, 500 insertions(+), 8 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 737c1c49bb4..7154b0b4c82 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
@@ -1439,4 +1439,11 @@ public final class 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.";
+ 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 fe569097189..973241cd431 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
@@ -1524,4 +1524,10 @@ public final class 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
=
+ "流程 {} 已越过提交点，保留它执行过的所有状态。";
+ 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/Procedure.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/Procedure.java
index 10ae561c2af..dc26ced3b64 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,24 @@ public abstract class Procedure<Env> implements
Comparable<Procedure<Env>> {
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 returns
+ * false is finished without undoing any of its states, see {@link
#abandonRollback()}.
+ */
+ protected boolean isRollbackSupported() {
+ 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 82b217b93ea..0b5460486b2 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.io.IOException;
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;
@@ -370,7 +372,7 @@ public class ProcedureExecutor<Env> {
*
* @param proc procedure
*/
- private void executeProcedure(Procedure<Env> proc) {
+ private void executeProcedure(Procedure<Env> proc) throws
InterruptedException {
if (proc.isFinished()) {
LOG.debug(ProcedureMessages.IS_ALREADY_FINISHED, proc);
return;
@@ -689,7 +691,7 @@ public class ProcedureExecutor<Env> {
* @return lock state
*/
private LockStateResult<Env> executeRootStackRollback(
- Long rootProcId, RootProcedureStack<Env> procedureStack) {
+ Long rootProcId, RootProcedureStack<Env> procedureStack) throws
InterruptedException {
Procedure<Env> rootProcedure = procedures.get(rootProcId);
ProcedureException exception = rootProcedure.getException();
if (exception == null) {
@@ -700,14 +702,33 @@ public class ProcedureExecutor<Env> {
() -> store.update(rootProcedure), "root procedure rollback");
}
List<Procedure<Env>> 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<Procedure<Env>> proceduresKeepingTheirStates =
+ Collections.newSetFromMap(new IdentityHashMap<>());
+ for (Procedure<Env> procedure : subprocStack) {
+ // 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<Env> procedure = subprocStack.get(stackTail);
+ if (procedure == null) {
+ continue;
+ }
if (procedure.isSuccess()) {
subprocStack.remove(stackTail);
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 +751,109 @@ public class ProcedureExecutor<Env> {
}
}
+ if (!proceduresKeepingTheirStates.isEmpty()) {
+ // 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<Env> procedure : proceduresKeepingTheirStates) {
+ if (procedure.hasParent()) {
+ finishWithoutRollback(procedure);
+ }
+ }
+ for (Procedure<Env> procedure : proceduresKeepingTheirStates) {
+ if (!procedure.hasParent()) {
+ finishWithoutRollback(procedure);
+ }
+ }
+ subprocStack.removeIf(proceduresKeepingTheirStates::contains);
+ }
+
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<Env> 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
+ // 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
+ 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.
+ // 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();
+ for (final long childProcId : childProcIds) {
+ 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);
+ }
+ 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.
+ 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(
+ ProcedureMessages
+
.LOG_PROCEDURE_ARG_IS_PAST_ITS_COMMIT_POINT_KEEPING_EVERY_STATE_IT_EXECUTED_CA21D7E9,
+ 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<Env> procedure) {
+ if (procedure.hasLock() || procedure.isLockedWhenLoading()) {
+ if (acquireLock(procedure) == ProcedureLockState.LOCK_ACQUIRED) {
+ releaseLock(procedure, true);
+ }
+ }
+ }
+
private ProcedureLockState acquireLock(Procedure<Env> 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 1fa240997f7..db38c6175bb 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 @@ public class RootProcedureStack<Env> {
protected synchronized ProcedureException getException() {
if (subprocStack != null) {
for (Procedure<Env> 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();
}
}
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 f34225cede9..5f518f728e9 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.DataOutputStream;
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;
@@ -252,12 +253,49 @@ public abstract class StateMachineProcedure<Env, TState>
extends Procedure<Env>
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.
+ * 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.
+ *
+ * <p>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<Integer> 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 23b041c2587..a6fdd7d6151 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 @@ public class RemoveAINodeProcedure extends
AbstractNodeProcedure<RemoveAINodeSta
// 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 be6f67007a3..a354f0ff3a3 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,23 @@ public abstract class AbstractSetPropertiesProcedure
protected abstract ConfigPhysicalPlan createSetPropertiesPlan(
Map<String, String> 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. The procedure
therefore stays
+ * rollbackable up to SET_PROPERTIES and not after it.
+ */
+ @Override
+ protected boolean isRollbackSupported(final SetTablePropertiesState state) {
+ switch (state) {
+ case VALIDATE_TABLE:
+ case PRE_RELEASE:
+ case SET_PROPERTIES:
+ return true;
+ default:
+ return false;
+ }
+ }
+
@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 44f9e2f07f9..07bdbf44c55 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,23 @@ public class AddTableColumnProcedure
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. The
procedure therefore
+ * stays rollbackable up to ADD_COLUMN and not after it.
+ */
+ @Override
+ protected boolean isRollbackSupported(final AddTableColumnState state) {
+ switch (state) {
+ case COLUMN_CHECK:
+ case PRE_RELEASE:
+ case ADD_COLUMN:
+ return true;
+ default:
+ return false;
+ }
+ }
+
@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 12491f24a48..1cfda63fcf7 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,7 @@ public class AlterTableColumnDataTypeProcedure
setNextState(AlterTableColumnDataTypeState.COMMIT_RELEASE);
}
+ /** 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 f2379ae1256..ac573463ebb 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
@@ -212,7 +212,15 @@ public class CreateTableProcedure
@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 a51c636551e..75000e36ef6 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,7 @@ public class DropTableColumnProcedure
}
}
+ /** 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 cdb4d4b71c8..54217212a11 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,9 +386,22 @@ public class DropTableProcedure extends
AbstractAlterOrDropTableProcedure<DropTa
}
}
+ /**
+ * 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. 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
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 11d10901a35..03088935ce6 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,23 @@ public class RenameTableColumnProcedure
}
}
+ /**
+ * 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. The procedure
therefore stays
+ * rollbackable up to RENAME_COLUMN and not after it.
+ */
+ @Override
+ protected boolean isRollbackSupported(final RenameTableColumnState state) {
+ switch (state) {
+ case COLUMN_CHECK:
+ case PRE_RELEASE:
+ case RENAME_COLUMN:
+ return true;
+ default:
+ return false;
+ }
+ }
+
@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 e40b041d0f6..52962533049 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,23 @@ public class RenameTableProcedure extends
AbstractAlterOrDropTableProcedure<Rena
}
}
+ /**
+ * 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. The procedure
therefore stays
+ * rollbackable up to RENAME_TABLE and not after it.
+ */
+ @Override
+ protected boolean isRollbackSupported(final RenameTableState state) {
+ switch (state) {
+ case COLUMN_CHECK:
+ case PRE_RELEASE:
+ case RENAME_TABLE:
+ return true;
+ default:
+ return false;
+ }
+ }
+
@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 73a7732fc87..163150c0a09 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,23 @@ public class DropViewProcedure extends DropTableProcedure {
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; before that it
can still be
+ * restored, so the procedure is not rolled back from COMMIT_DELETE on.
+ */
+ @Override
+ 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
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 00000000000..340f1b88c5c
--- /dev/null
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/StateMachineProcedureTest.java
@@ -0,0 +1,185 @@
+/*
+ * 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 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<TestProcEnv, State> {
+
+ private boolean failAtApply;
+ private boolean failAtRelease;
+
+ @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 {
+ setNextState(State.RELEASE);
+ }
+ break;
+ case RELEASE:
+ if (failAtRelease) {
+ setFailure(new ProcedureException("release failed"));
+ }
+ 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 called once the commit
point has been passed.
+ }
+
+ @Override
+ protected boolean isRollbackSupported(final State state) {
+ return state != State.RELEASE;
+ }
+
+ @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 State currentState() {
+ return getCurrentState();
+ }
+
+ private void addSlot(final int index) {
+ addStackIndex(index);
+ }
+
+ private int[] stackIndexes() {
+ return getStackIndexes();
+ }
+
+ private void abandon() {
+ abandonRollback();
+ }
+ }
+
+ @Test
+ public void rollbackIsAllowedUntilTheStateAfterTheCommitPointIsReached()
+ throws InterruptedException {
+ final CommitPointProcedure procedure = new CommitPointProcedure();
+
+ // PREPARE has completed and APPLY is scheduled.
+ procedure.runStep();
+ Assert.assertEquals(State.APPLY, procedure.currentState());
+ Assert.assertTrue(procedure.isRollbackSupported());
+
+ // APPLY has completed: the procedure is at RELEASE and must not be rolled
back any more.
+ procedure.runStep();
+ Assert.assertEquals(State.RELEASE, procedure.currentState());
+ Assert.assertFalse(procedure.isRollbackSupported());
+
+ // Every state has run; the decision falls on the last state which ran,
not on the end marker.
+ procedure.runStep();
+ Assert.assertFalse(procedure.isRollbackSupported());
+ }
+
+ @Test
+ public void aFailureBeforeTheCommitPointStillAllowsTheRollback() throws
InterruptedException {
+ final CommitPointProcedure procedure = new CommitPointProcedure();
+ procedure.runStep();
+
+ procedure.failAtApply = true;
+ procedure.runStep();
+
+ Assert.assertEquals(State.APPLY, procedure.currentState());
+ Assert.assertTrue(procedure.isRollbackSupported());
+ }
+
+ @Test
+ public void aFailureAfterTheCommitPointDoesNotAllowTheRollback() throws
InterruptedException {
+ final CommitPointProcedure procedure = new CommitPointProcedure();
+ procedure.runStep();
+ procedure.runStep();
+
+ procedure.failAtRelease = true;
+ procedure.runStep();
+
+ Assert.assertFalse(procedure.isRollbackSupported());
+ }
+
+ @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.isRollbackSupported());
+
+ 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 aProcedureWhichOverridesNothingIsRollbackableEverywhere() {
+ Assert.assertTrue(new SimpleSTMProcedure().isRollbackSupported());
+ }
+}
