Fix StateMachineProcedure EOF state re-execution

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({})";

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({})";

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/StateMachineProcedure.java

@@ -146,10 +146,17 @@ protected Procedure<Env>[] execute(final Env env) throws InterruptedException {
     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 @@ protected Procedure<Env>[] execute(final Env env) throws InterruptedException {
         subProcList.clear();
         return subProcedures;
       }
-      return (isWaiting() || isFailed() || noMoreState()) ? null : new Procedure[] {this};
+      return (isWaiting() || isFailed() || noMoreState())
+          ? new Procedure[0]
+          : new Procedure[] {this};
     } finally {
       updateTimestamp();
     }

iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/STMProcedureTest.java

@@ -21,11 +21,15 @@
 
 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 void testRolledBackProcedure() {
     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;
+    }
+  }
 }

iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2Test.java

@@ -82,7 +82,8 @@ public void testRetryStateYieldsOnlyBeforeRetryThreshold() throws Exception {
     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);