Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -145,10 +145,19 @@ protected Procedure<Env>[] 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()) {
Expand All @@ -168,7 +177,9 @@ protected Procedure<Env>[] 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();
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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<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;
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
Loading