From 05b9dca98b96819748545d5232496666722c9fd5 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Mon, 6 Jul 2026 12:08:04 +0800 Subject: [PATCH] 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 (cherry picked from commit 8d739903ede9cb274771825a18de96f974c5d23b) --- .../procedure/impl/StateMachineProcedure.java | 15 +++- .../procedure/STMProcedureTest.java | 70 +++++++++++++++++++ .../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 400da5653181c..588a04dacb191 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 @@ protected Procedure[] 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( + "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 @@ protected Procedure[] 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(); } 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 3ccbb2e15162d..1f62a5d1aaae3 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 @@ 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 states = + (ConcurrentLinkedDeque) 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 { + + 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 be6b6374e8572..5993e73ec42a9 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 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);