This is an automated email from the ASF dual-hosted git repository.
jt2594838 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 8d739903ede Fix StateMachineProcedure EOF state re-execution (#18099)
8d739903ede is described below
commit 8d739903ede9cb274771825a18de96f974c5d23b
Author: Caideyipi <[email protected]>
AuthorDate: Mon Jul 6 12:08:04 2026 +0800
Fix StateMachineProcedure EOF state re-execution (#18099)
* Fix state machine EOF state reexecution
* Add i18n for EOF procedure log
* Fix StateMachineProcedure formatting
* Avoid null procedure arrays in state machine
* Align pipe procedure tests with empty child result
---
.../iotdb/confignode/i18n/ProcedureMessages.java | 2 +
.../iotdb/confignode/i18n/ProcedureMessages.java | 2 +
.../procedure/impl/StateMachineProcedure.java | 13 +++-
.../confignode/procedure/STMProcedureTest.java | 70 ++++++++++++++++++++++
.../pipe/AbstractOperatePipeProcedureV2Test.java | 3 +-
5 files changed, 87 insertions(+), 3 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 3d557371219..02fddae7d6a 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
@@ -915,6 +915,8 @@ public final class ProcedureMessages {
"Start rollback Renaming table {}.{} on configNode";
public static final String START_ROLLBACK_SET_PROPERTIES_TO_TABLE =
"Start rollback set properties to table {}.{}";
+ public static final String STATE_MACHINE_PROCEDURE_EOF_STATE_SKIP_EXECUTION =
+ "StateMachineProcedure pid={} is scheduled with EOF state, skip
execution: {}";
public static final String STATE_STUCK_AT = "State stuck at ";
public static final String
STOPPIPEPROCEDUREV2_EXECUTEFROMCALCULATEINFOFORTASK =
"StopPipeProcedureV2: executeFromCalculateInfoForTask({})";
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 4dc51d4d79a..e2a5ca86039 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
@@ -913,6 +913,8 @@ public final class ProcedureMessages {
"在 ConfigNode 上开始回滚重命名表 {}.{}";
public static final String START_ROLLBACK_SET_PROPERTIES_TO_TABLE =
"开始回滚设置表 {}.{} 的属性";
+ public static final String STATE_MACHINE_PROCEDURE_EOF_STATE_SKIP_EXECUTION =
+ "StateMachineProcedure pid={} 被 EOF 状态调度，跳过执行：{}";
public static final String STATE_STUCK_AT = "状态卡在 ";
public static final String
STOPPIPEPROCEDUREV2_EXECUTEFROMCALCULATEINFOFORTASK =
"StopPipeProcedureV2: executeFromCalculateInfoForTask({})";
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 07b0d5e1c5b..8f3c1c93633 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
@@ -146,10 +146,17 @@ public abstract class StateMachineProcedure<Env, TState>
extends Procedure<Env>
updateTimestamp();
try {
if (noMoreState() || isFailed()) {
- return null;
+ return new Procedure[0];
}
TState state = getCurrentState();
+ if (state == null) {
+ LOG.warn(
+
ProcedureMessages.STATE_MACHINE_PROCEDURE_EOF_STATE_SKIP_EXECUTION,
getProcId(), this);
+ stateFlow = Flow.NO_MORE_STATE;
+ setStateDeserialized(false);
+ return new Procedure[0];
+ }
// init for the first execution
if (states.isEmpty()) {
@@ -169,7 +176,9 @@ public abstract class StateMachineProcedure<Env, TState>
extends Procedure<Env>
subProcList.clear();
return subProcedures;
}
- return (isWaiting() || isFailed() || noMoreState()) ? null : new
Procedure[] {this};
+ return (isWaiting() || isFailed() || noMoreState())
+ ? new Procedure[0]
+ : new Procedure[] {this};
} finally {
updateTimestamp();
}
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/STMProcedureTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/STMProcedureTest.java
index 3ccbb2e1516..1f62a5d1aaa 100644
---
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/STMProcedureTest.java
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/STMProcedureTest.java
@@ -21,11 +21,15 @@ package org.apache.iotdb.confignode.procedure;
import org.apache.iotdb.confignode.procedure.entity.SimpleSTMProcedure;
import org.apache.iotdb.confignode.procedure.env.TestProcEnv;
+import org.apache.iotdb.confignode.procedure.impl.StateMachineProcedure;
+import org.apache.iotdb.confignode.procedure.state.ProcedureState;
import org.apache.iotdb.confignode.procedure.util.ProcedureTestUtil;
import org.junit.Assert;
import org.junit.Test;
+import java.lang.reflect.Field;
+import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.atomic.AtomicInteger;
public class STMProcedureTest extends TestProcedureBase {
@@ -55,4 +59,70 @@ public class STMProcedureTest extends TestProcedureBase {
System.out.println(rolledback);
Assert.assertEquals(1 + success - rolledback, acc.get());
}
+
+ @Test
+ public void testEofStateReexecutionDoesNotCallExecuteFromState() throws
Exception {
+ EofReexecutionProcedure procedure = new EofReexecutionProcedure();
+ procedure.setProcId(1);
+ procedure.setState(ProcedureState.RUNNABLE);
+
+ forceEofStateWithHasMoreFlow(procedure);
+
+ Assert.assertEquals(0, procedure.doExecute(env).length);
+ Assert.assertEquals(0, procedure.executeCount);
+ }
+
+ private static void forceEofStateWithHasMoreFlow(StateMachineProcedure<?, ?>
procedure)
+ throws Exception {
+ Field eofStateField =
StateMachineProcedure.class.getDeclaredField("EOF_STATE");
+ eofStateField.setAccessible(true);
+
+ Field statesField = StateMachineProcedure.class.getDeclaredField("states");
+ statesField.setAccessible(true);
+ @SuppressWarnings("unchecked")
+ ConcurrentLinkedDeque<Integer> states =
+ (ConcurrentLinkedDeque<Integer>) statesField.get(procedure);
+ states.clear();
+ states.add(eofStateField.getInt(null));
+
+ Field stateFlowField =
StateMachineProcedure.class.getDeclaredField("stateFlow");
+ stateFlowField.setAccessible(true);
+ stateFlowField.set(procedure, StateMachineProcedure.Flow.HAS_MORE_STATE);
+ }
+
+ private static class EofReexecutionProcedure
+ extends StateMachineProcedure<TestProcEnv,
EofReexecutionProcedure.TestState> {
+
+ private int executeCount = 0;
+
+ private enum TestState {
+ STEP
+ }
+
+ @Override
+ protected Flow executeFromState(TestProcEnv testProcEnv, TestState
testState) {
+ executeCount++;
+ return Flow.NO_MORE_STATE;
+ }
+
+ @Override
+ protected void rollbackState(TestProcEnv testProcEnv, TestState testState)
{
+ // No rollback work is required for this regression test.
+ }
+
+ @Override
+ protected TestState getState(int stateId) {
+ return TestState.values()[stateId];
+ }
+
+ @Override
+ protected int getStateId(TestState testState) {
+ return testState.ordinal();
+ }
+
+ @Override
+ protected TestState getInitialState() {
+ return TestState.STEP;
+ }
+ }
}
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2Test.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2Test.java
index be6b6374e85..5993e73ec42 100644
---
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2Test.java
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2Test.java
@@ -82,7 +82,8 @@ public class AbstractOperatePipeProcedureV2Test {
Assert.assertTrue(procedure.isYieldAfterExecution(null));
Assert.assertEquals(1, procedure.calculateExecutionCount);
- Assert.assertNull(procedure.runOnce());
+ final Procedure<?>[] failedSubProcedures = procedure.runOnce();
+ Assert.assertEquals(0, failedSubProcedures.length);
Assert.assertTrue(procedure.hasException());
Assert.assertFalse(procedure.isYieldAfterExecution(null));
Assert.assertEquals(2, procedure.calculateExecutionCount);
