This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new c7c54941e32 Fix StateMachineProcedure EOF state re-execution (#18099)
(#18122)
c7c54941e32 is described below
commit c7c54941e32ac0752d055cd00f90d3d43b2ca988
Author: Caideyipi <[email protected]>
AuthorDate: Tue Jul 7 10:33:09 2026 +0800
Fix StateMachineProcedure EOF state re-execution (#18099) (#18122)
* 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
(cherry picked from commit 8d739903ede9cb274771825a18de96f974c5d23b)
---
.../procedure/impl/StateMachineProcedure.java | 15 ++++-
.../confignode/procedure/STMProcedureTest.java | 70 ++++++++++++++++++++++
.../pipe/AbstractOperatePipeProcedureV2Test.java | 3 +-
3 files changed, 85 insertions(+), 3 deletions(-)
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 400da565318..588a04dacb1 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
@@ -145,10 +145,19 @@ 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(
+ "StateMachineProcedure pid={} is scheduled with 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()) {
@@ -168,7 +177,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);
