(iotdb) branch fix/pipe-procedure-eof-state-npe created (now 5a3c6d305ef)

[ju...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

Caideyipi pushed a change to branch fix/pipe-procedure-eof-state-npe
in repository https://gitbox.apache.org/repos/asf/iotdb.git


      at 5a3c6d305ef Fix state machine EOF state reexecution

This branch includes the following new commits:

     new 5a3c6d305ef Fix state machine EOF state reexecution

The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails.  The revisions
listed as "add" were already present in the repository and have only
been added to this reference.

(iotdb) 01/01: Fix state machine EOF state reexecution

[ju...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

Caideyipi pushed a commit to branch fix/pipe-procedure-eof-state-npe
in repository https://gitbox.apache.org/repos/asf/iotdb.git

commit 5a3c6d305eff760bb796acda20bdba194bc512b0
Author: Caideyipi <87...@users.noreply.github.com>
AuthorDate: Fri Jul 3 11:17:11 2026 +0800

    Fix state machine EOF state reexecution
---
 .../procedure/impl/StateMachineProcedure.java      |  9 +++
 .../confignode/procedure/STMProcedureTest.java     | 70 ++++++++++++++++++++++
 2 files changed, 79 insertions(+)

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 f7275a152d7..d4b4fb61fc8 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
@@ -149,6 +149,15 @@ public abstract class StateMachineProcedure<Env, TState> extends Procedure<Env>
       }
 
       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 null;
+      }
 
       // init for the first execution
       if (states.isEmpty()) {
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..fe342d82a33 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.assertNull(procedure.doExecute(env));
+    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;
+    }
+  }
 }