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 <[email protected]> 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; + } + } }
