From 5a3c6d305eff760bb796acda20bdba194bc512b0 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 3 Jul 2026 11:17:11 +0800 Subject: [PATCH 1/5] Fix state machine EOF state reexecution --- .../procedure/impl/StateMachineProcedure.java | 9 +++ .../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 f7275a152d7e7..d4b4fb61fc864 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 @@ protected Procedure[] execute(final Env env) throws InterruptedException { } 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 3ccbb2e15162d..fe342d82a33a8 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.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 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; + } + } } From 52632c4ee8e114cf6b65fefd662aaec48349dcb0 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 3 Jul 2026 13:04:42 +0800 Subject: [PATCH 2/5] Add i18n for EOF procedure log --- .../en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java | 2 ++ .../zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java | 2 ++ .../iotdb/confignode/procedure/impl/StateMachineProcedure.java | 3 ++- 3 files changed, 6 insertions(+), 1 deletion(-) 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 87e18f5ba906d..58131ffc9f969 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 1cb059d6c72f8..d95056e6f2b6a 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 { "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={} 被 EOF 状态调度，跳过执行：{}"; 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/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 d4b4fb61fc864..0474a025d33ef 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 @@ -19,6 +19,7 @@ package org.apache.iotdb.confignode.procedure.impl; +import org.apache.iotdb.confignode.i18n.ProcedureMessages; import org.apache.iotdb.confignode.procedure.Procedure; import org.apache.iotdb.confignode.procedure.exception.ProcedureException; @@ -151,7 +152,7 @@ protected Procedure[] execute(final Env env) throws InterruptedException { TState state = getCurrentState(); if (state == null) { LOG.warn( - "StateMachineProcedure pid={} is scheduled with EOF state, skip execution: {}", + ProcedureMessages.STATE_MACHINE_PROCEDURE_EOF_STATE_SKIP_EXECUTION, getProcId(), this); stateFlow = Flow.NO_MORE_STATE; From 0670b671e7a20ec081b1b4103f7d58e0173a14e7 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 3 Jul 2026 15:37:21 +0800 Subject: [PATCH 3/5] Fix StateMachineProcedure formatting --- .../confignode/procedure/impl/StateMachineProcedure.java | 4 +--- 1 file changed, 1 insertion(+), 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 0474a025d33ef..91da1b05f6d4e 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 @@ -152,9 +152,7 @@ protected Procedure[] execute(final Env env) throws InterruptedException { TState state = getCurrentState(); if (state == null) { LOG.warn( - ProcedureMessages.STATE_MACHINE_PROCEDURE_EOF_STATE_SKIP_EXECUTION, - getProcId(), - this); + ProcedureMessages.STATE_MACHINE_PROCEDURE_EOF_STATE_SKIP_EXECUTION, getProcId(), this); stateFlow = Flow.NO_MORE_STATE; setStateDeserialized(false); return null; From cbab7377475cf83d292042c155f9e619edbcc1de Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 3 Jul 2026 16:31:23 +0800 Subject: [PATCH 4/5] Avoid null procedure arrays in state machine --- .../confignode/procedure/impl/StateMachineProcedure.java | 8 +++++--- 1 file changed, 5 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 91da1b05f6d4e..53785ec5349fe 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,7 +146,7 @@ protected Procedure[] execute(final Env env) throws InterruptedException { updateTimestamp(); try { if (noMoreState() || isFailed()) { - return null; + return new Procedure[0]; } TState state = getCurrentState(); @@ -155,7 +155,7 @@ protected Procedure[] execute(final Env env) throws InterruptedException { ProcedureMessages.STATE_MACHINE_PROCEDURE_EOF_STATE_SKIP_EXECUTION, getProcId(), this); stateFlow = Flow.NO_MORE_STATE; setStateDeserialized(false); - return null; + return new Procedure[0]; } // init for the first execution @@ -176,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(); } From 96f883c85a7f00b57ce5663e3afdcc915123a26f Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 3 Jul 2026 19:13:31 +0800 Subject: [PATCH 5/5] Align pipe procedure tests with empty child result --- .../apache/iotdb/confignode/procedure/STMProcedureTest.java | 2 +- .../impl/pipe/AbstractOperatePipeProcedureV2Test.java | 3 ++- 2 files changed, 3 insertions(+), 2 deletions(-) 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 fe342d82a33a8..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 @@ -68,7 +68,7 @@ public void testEofStateReexecutionDoesNotCallExecuteFromState() throws Exceptio forceEofStateWithHasMoreFlow(procedure); - Assert.assertNull(procedure.doExecute(env)); + Assert.assertEquals(0, procedure.doExecute(env).length); Assert.assertEquals(0, procedure.executeCount); } 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);