diff --git a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java index 3d5573712197..02fddae7d6a2 100644 --- a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java +++ b/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({})"; diff --git a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java index 4dc51d4d79a1..e2a5ca860398 100644 --- a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java +++ b/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({})"; 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 07b0d5e1c5b5..8f3c1c93633c 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 @@ -146,10 +146,17 @@ 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( + 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[] 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);