This is an automated email from the ASF dual-hosted git repository.
CRZbulabula pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 32bff9b615a Fix procedures stuck after submission-time deserialization
failure (#18212)
32bff9b615a is described below
commit 32bff9b615a68e1a8fd79320415591a45508ca1e
Author: Yongzao <[email protected]>
AuthorDate: Wed Jul 15 16:36:16 2026 +0800
Fix procedures stuck after submission-time deserialization failure (#18212)
---
.../confignode/procedure/ProcedureExecutor.java | 28 +++++
.../procedure/TestProcedureExecutor.java | 127 +++++++++++++++++++++
2 files changed, 155 insertions(+)
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java
index 92ce0b9c2c9..2aca885c391 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java
@@ -211,6 +211,14 @@ public class ProcedureExecutor<Env> {
runnableList.add(procedure);
}
});
+ // A submission-time deserialization failure may be persisted before a
rollback stack index.
+ failedList.forEach(
+ procedure -> {
+ RootProcedureStack<Env> rootStack =
rollbackStack.get(getRootProcedureId(procedure));
+ if (rootStack != null) {
+ initializeRollbackStackForFailedProcedure(rootStack, procedure);
+ }
+ });
restoreLocks();
waitingTimeoutList.forEach(timeoutExecutor::add);
@@ -810,12 +818,32 @@ public class ProcedureExecutor<Env> {
// Update metrics on start of a procedure
procedure.updateMetricsOnSubmit(getEnvironment());
RootProcedureStack<Env> stack = new RootProcedureStack<>();
+ // Persisting a newly submitted procedure may serialize and deserialize it
through the
+ // consensus layer. If that process marks the procedure as failed before
it is scheduled, the
+ // rollback stack still needs an entry so the executor can finish the
failed procedure instead
+ // of leaving it in the active procedure map forever.
+ if (initializeRollbackStackForFailedProcedure(stack, procedure)) {
+ try {
+ store.update(procedure);
+ } catch (Exception e) {
+ LOG.error(ProcedureMessages.FAILED_TO_UPDATE_STORE_PROCEDURE,
procedure, e);
+ }
+ }
rollbackStack.put(currentProcId, stack);
procedures.put(currentProcId, procedure);
scheduler.addBack(procedure);
return procedure.getProcId();
}
+ private boolean initializeRollbackStackForFailedProcedure(
+ RootProcedureStack<Env> stack, Procedure<Env> procedure) {
+ if (procedure.isFailed() && !procedure.wasExecuted()) {
+ stack.addRollbackStep(procedure);
+ return true;
+ }
+ return false;
+ }
+
private class WorkerThread extends StoppableThread {
private final AtomicLong startTime = new AtomicLong(Long.MAX_VALUE);
private final AtomicReference<Procedure<Env>> activeProcedure = new
AtomicReference<>();
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/TestProcedureExecutor.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/TestProcedureExecutor.java
index ab6ec55f4d7..95caffd4a9a 100644
---
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/TestProcedureExecutor.java
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/TestProcedureExecutor.java
@@ -31,8 +31,11 @@ import org.junit.Assert;
import org.junit.Test;
import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
public class TestProcedureExecutor extends TestProcedureBase {
@@ -56,6 +59,66 @@ public class TestProcedureExecutor extends TestProcedureBase
{
Assert.assertEquals(acc.get(), 1);
}
+ @Test
+ public void testProcedureFailedDuringSubmissionIsRolledBack() throws
InterruptedException {
+ TestProcEnv localEnv = new TestProcEnv();
+ FailOnFirstUpdateProcedureStore localStore = new
FailOnFirstUpdateProcedureStore();
+ ProcedureExecutor<TestProcEnv> localExecutor = new
ProcedureExecutor<>(localEnv, localStore);
+ localEnv.setScheduler(localExecutor.getScheduler());
+ localStore.start();
+ localExecutor.init(1);
+
+ try {
+ FailedDuringSubmissionProcedure procedure = new
FailedDuringSubmissionProcedure();
+ long procId = localExecutor.submitProcedure(procedure);
+
+ boolean rollbackStackPersisted = localStore.isRollbackStackPersisted();
+ localExecutor.startWorkers();
+ Assert.assertTrue(rollbackStackPersisted);
+ Assert.assertTrue(procedure.awaitRollback(10, TimeUnit.SECONDS));
+ ProcedureTestUtil.waitForProcedure(localExecutor, procId);
+ Assert.assertTrue(localExecutor.isFinished(procId));
+ Assert.assertEquals(ProcedureState.ROLLEDBACK, procedure.getState());
+ Assert.assertEquals(0, procedure.getExecutionCount());
+ Assert.assertEquals(1, procedure.getRollbackCount());
+ } finally {
+ localStore.stop();
+ localExecutor.stop();
+ localExecutor.join();
+ localStore.cleanup();
+ }
+ }
+
+ @Test
+ public void testFailedProcedureWithoutRollbackStackCanBeRecovered() throws
InterruptedException {
+ FailedDuringSubmissionProcedure procedure = new
FailedDuringSubmissionProcedure();
+ procedure.setProcId(0);
+ procedure.setProcRunnable();
+ procedure.setFailure(new ProcedureException("Failed to deserialize
procedure"));
+
+ TestProcEnv localEnv = new TestProcEnv();
+ FailedProcedureStore localStore = new FailedProcedureStore(procedure);
+ ProcedureExecutor<TestProcEnv> localExecutor = new
ProcedureExecutor<>(localEnv, localStore);
+ localEnv.setScheduler(localExecutor.getScheduler());
+ localStore.start();
+ localExecutor.init(1);
+ localExecutor.startWorkers();
+
+ try {
+ Assert.assertTrue(procedure.awaitRollback(10, TimeUnit.SECONDS));
+ ProcedureTestUtil.waitForProcedure(localExecutor, procedure.getProcId());
+ Assert.assertTrue(localExecutor.isFinished(procedure.getProcId()));
+ Assert.assertEquals(ProcedureState.ROLLEDBACK, procedure.getState());
+ Assert.assertEquals(0, procedure.getExecutionCount());
+ Assert.assertEquals(1, procedure.getRollbackCount());
+ } finally {
+ localStore.stop();
+ localExecutor.stop();
+ localExecutor.join();
+ localStore.cleanup();
+ }
+ }
+
@Test
public void testWorkerThreadStuck() throws InterruptedException {
Semaphore latch1 = new Semaphore(2);
@@ -223,4 +286,68 @@ public class TestProcedureExecutor extends
TestProcedureBase {
return executionCount.get();
}
}
+
+ private static class FailOnFirstUpdateProcedureStore extends
NoopProcedureStore {
+
+ private final AtomicBoolean firstUpdate = new AtomicBoolean(true);
+ private final AtomicBoolean rollbackStackPersisted = new
AtomicBoolean(false);
+
+ @Override
+ public void update(Procedure procedure) {
+ if (firstUpdate.compareAndSet(true, false)) {
+ procedure.setFailure(new ProcedureException("Failed to deserialize
procedure"));
+ } else if (procedure.isFailed() && procedure.wasExecuted()) {
+ rollbackStackPersisted.set(true);
+ }
+ }
+
+ private boolean isRollbackStackPersisted() {
+ return rollbackStackPersisted.get();
+ }
+ }
+
+ private static class FailedProcedureStore extends NoopProcedureStore {
+
+ private final Procedure procedure;
+
+ private FailedProcedureStore(Procedure procedure) {
+ this.procedure = procedure;
+ }
+
+ @Override
+ public List<Procedure> load() {
+ return Collections.singletonList(procedure);
+ }
+ }
+
+ private static class FailedDuringSubmissionProcedure extends
Procedure<TestProcEnv> {
+
+ private final Semaphore rolledBack = new Semaphore(0);
+ private final AtomicInteger executionCount = new AtomicInteger();
+ private final AtomicInteger rollbackCount = new AtomicInteger();
+
+ @Override
+ protected Procedure<TestProcEnv>[] execute(TestProcEnv env) {
+ executionCount.incrementAndGet();
+ return null;
+ }
+
+ @Override
+ protected void rollback(TestProcEnv env) {
+ rollbackCount.incrementAndGet();
+ rolledBack.release();
+ }
+
+ private boolean awaitRollback(long timeout, TimeUnit unit) throws
InterruptedException {
+ return rolledBack.tryAcquire(timeout, unit);
+ }
+
+ private int getExecutionCount() {
+ return executionCount.get();
+ }
+
+ private int getRollbackCount() {
+ return rollbackCount.get();
+ }
+ }
}
