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 400da5653181..588a04dacb19 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 3ccbb2e15162..1f62a5d1aaae 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 be6b6374e857..5993e73ec42a 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);