From 62a76dfa1e3f37aa5058519b6725037154a86222 Mon Sep 17 00:00:00 2001 From: Di Wang Date: Sat, 24 Oct 2026 01:27:16 +0800 Subject: [PATCH] Persist parent and child procedures atomically before scheduling Commit parent progress and new children in one ConfigNode consensus entry. Retry failed writes before scheduling and preserve recovery/snapshot consistency. Add regression coverage for restart, retry, interruption, and serialization. --- .../confignode/i18n/ProcedureMessages.java | 2 + .../confignode/i18n/ProcedureMessages.java | 1 + .../consensus/request/ConfigPhysicalPlan.java | 4 + .../request/ConfigPhysicalPlanType.java | 1 + .../write/procedure/UpdateProceduresPlan.java | 93 ++++++ .../confignode/persistence/ProcedureInfo.java | 14 +- .../executor/ConfigPlanExecutor.java | 3 + .../procedure/ProcedureExecutor.java | 17 +- .../procedure/store/ConfigProcedureStore.java | 19 +- .../procedure/store/IProcedureStore.java | 3 +- .../request/UpdateProceduresPlanTest.java | 114 ++++++++ .../ProcedureChildPersistenceTest.java | 266 ++++++++++++++++++ 12 files changed, 523 insertions(+), 14 deletions(-) create mode 100644 iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/procedure/UpdateProceduresPlan.java create mode 100644 iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/UpdateProceduresPlanTest.java create mode 100644 iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/ProcedureChildPersistenceTest.java 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 488e7cc759841..5263182f2dc2d 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 @@ -1453,4 +1453,6 @@ private ProcedureMessages() {} "finish procedure without rollback"; public static final String LOG_DELETE_CHILD_PROCEDURES_WITHOUT_ROLLBACK_6E62D1C1 = "delete child procedures without rollback"; + public static final String LOG_PERSIST_PARENT_AND_CHILD_PROCEDURES_54E2D0D2 = + "persist parent and child procedures"; } 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 d65ee58ec53a7..f27c77a07c5a8 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 @@ -1536,4 +1536,5 @@ private ProcedureMessages() {} public static final String LOG_FINISH_PROCEDURE_WITHOUT_ROLLBACK_41F7A354 = "结束流程（不回滚）"; public static final String LOG_DELETE_CHILD_PROCEDURES_WITHOUT_ROLLBACK_6E62D1C1 = "删除子流程（不回滚）"; + public static final String LOG_PERSIST_PARENT_AND_CHILD_PROCEDURES_54E2D0D2 = "持久化父流程和子流程"; } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlan.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlan.java index 9c01288d287f6..0f7a90544e2df 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlan.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlan.java @@ -84,6 +84,7 @@ import org.apache.iotdb.confignode.consensus.request.write.pipe.task.SetPipeStatusWithStoppedByRuntimeExceptionPlanV2; import org.apache.iotdb.confignode.consensus.request.write.procedure.DeleteProcedurePlan; import org.apache.iotdb.confignode.consensus.request.write.procedure.UpdateProcedurePlan; +import org.apache.iotdb.confignode.consensus.request.write.procedure.UpdateProceduresPlan; import org.apache.iotdb.confignode.consensus.request.write.quota.SetSpaceQuotaPlan; import org.apache.iotdb.confignode.consensus.request.write.quota.SetThrottleQuotaPlan; import org.apache.iotdb.confignode.consensus.request.write.region.CreateRegionGroupsPlan; @@ -280,6 +281,9 @@ public static ConfigPhysicalPlan create(final ByteBuffer buffer) throws IOExcept case DeleteProcedure: plan = new DeleteProcedurePlan(); break; + case UpdateProcedures: + plan = new UpdateProceduresPlan(); + break; case UpdateProcedure: plan = new UpdateProcedurePlan(); break; diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanType.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanType.java index c1d8878ee9bd9..d9db2c0c0a9a9 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanType.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanType.java @@ -88,6 +88,7 @@ public enum ConfigPhysicalPlanType { /** Procedure. */ UpdateProcedure((short) 500), DeleteProcedure((short) 501), + UpdateProcedures((short) 502), /** Authority. */ Author((short) 600), diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/procedure/UpdateProceduresPlan.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/procedure/UpdateProceduresPlan.java new file mode 100644 index 0000000000000..06750e84c7481 --- /dev/null +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/procedure/UpdateProceduresPlan.java @@ -0,0 +1,93 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.confignode.consensus.request.write.procedure; + +import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan; +import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanType; +import org.apache.iotdb.confignode.procedure.Procedure; +import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv; + +import java.io.DataOutputStream; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.List; +import java.util.Objects; + +/** A single consensus entry for a parent procedure and its newly submitted children. */ +public class UpdateProceduresPlan extends ConfigPhysicalPlan { + private List> procedures; + + public UpdateProceduresPlan() { + super(ConfigPhysicalPlanType.UpdateProcedures); + } + + public UpdateProceduresPlan(List> procedures) { + this(); + this.procedures = new ArrayList<>(procedures); + } + + public List> getProcedures() { + return procedures; + } + + @Override + protected void serializeImpl(DataOutputStream stream) throws IOException { + stream.writeShort(getType().getPlanType()); + stream.writeInt(procedures.size()); + for (Procedure procedure : procedures) { + ByteBuffer buffer = new UpdateProcedurePlan(procedure).serializeToByteBuffer(); + stream.writeInt(buffer.remaining()); + stream.write(buffer.array(), buffer.position(), buffer.remaining()); + } + } + + @Override + protected void deserializeImpl(ByteBuffer buffer) throws IOException { + int size = buffer.getInt(); + procedures = new ArrayList<>(size); + for (int i = 0; i < size; i++) { + int length = buffer.getInt(); + // Some procedures read optional trailing fields. Isolate each record from the next one. + ByteBuffer procedureBuffer = buffer.slice(); + procedureBuffer.limit(length); + procedures.add( + ((UpdateProcedurePlan) ConfigPhysicalPlan.Factory.create(procedureBuffer)) + .getProcedure()); + buffer.position(buffer.position() + length); + } + } + + @Override + public boolean equals(Object obj) { + if (this == obj) { + return true; + } + if (obj == null || getClass() != obj.getClass()) { + return false; + } + return procedures.equals(((UpdateProceduresPlan) obj).procedures); + } + + @Override + public int hashCode() { + return Objects.hash(procedures); + } +} diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/ProcedureInfo.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/ProcedureInfo.java index 7409258b8c9fa..a7a864e51af7d 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/ProcedureInfo.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/ProcedureInfo.java @@ -27,6 +27,7 @@ import org.apache.iotdb.commons.utils.TestOnly; import org.apache.iotdb.confignode.consensus.request.write.procedure.DeleteProcedurePlan; import org.apache.iotdb.confignode.consensus.request.write.procedure.UpdateProcedurePlan; +import org.apache.iotdb.confignode.consensus.request.write.procedure.UpdateProceduresPlan; import org.apache.iotdb.confignode.i18n.ConfigNodeMessages; import org.apache.iotdb.confignode.manager.ConfigManager; import org.apache.iotdb.confignode.procedure.Procedure; @@ -142,6 +143,15 @@ public TSStatus updateProcedure(UpdateProcedurePlan updateProcedurePlan) { return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()); } + // Recovery and snapshots must not observe only part of a parent/children batch. + public synchronized TSStatus updateProcedures(UpdateProceduresPlan plan) { + for (Procedure procedure : plan.getProcedures()) { + procedureMap.put(procedure.getProcId(), procedure); + lastProcId.updateAndGet(id -> Math.max(id, procedure.getProcId())); + } + return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()); + } + @TestOnly public TSStatus oldUpdateProcedure(UpdateProcedurePlan updateProcedurePlan) { Procedure procedure = updateProcedurePlan.getProcedure(); @@ -194,7 +204,7 @@ private static Optional loadProcedure(Path procedureFilePath) { } @Override - public boolean processTakeSnapshot(File snapshotDir) throws TException, IOException { + public synchronized boolean processTakeSnapshot(File snapshotDir) throws TException, IOException { File procedureSnapshotDir = new File(snapshotDir, PROCEDURE_SNAPSHOT_DIR); if (procedureSnapshotDir.exists()) { LOGGER.error( @@ -275,7 +285,7 @@ public void processLoadSnapshot(File snapshotDir) throws TException, IOException }); } - public List> getProcedures() { + public synchronized List> getProcedures() { return new ArrayList<>(procedureMap.values()); } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigPlanExecutor.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigPlanExecutor.java index 8919308a159c5..9b428eeb775ef 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigPlanExecutor.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigPlanExecutor.java @@ -110,6 +110,7 @@ import org.apache.iotdb.confignode.consensus.request.write.pipe.task.SetPipeStatusWithStoppedByRuntimeExceptionPlanV2; import org.apache.iotdb.confignode.consensus.request.write.procedure.DeleteProcedurePlan; import org.apache.iotdb.confignode.consensus.request.write.procedure.UpdateProcedurePlan; +import org.apache.iotdb.confignode.consensus.request.write.procedure.UpdateProceduresPlan; import org.apache.iotdb.confignode.consensus.request.write.quota.SetSpaceQuotaPlan; import org.apache.iotdb.confignode.consensus.request.write.quota.SetThrottleQuotaPlan; import org.apache.iotdb.confignode.consensus.request.write.region.CreateRegionGroupsPlan; @@ -451,6 +452,8 @@ public TSStatus executeNonQueryPlan(ConfigPhysicalPlan physicalPlan) return partitionInfo.createDataPartition((CreateDataPartitionPlan) physicalPlan); case AutoCleanPartitionTable: return partitionInfo.autoCleanPartitionTable((AutoCleanPartitionTablePlan) physicalPlan); + case UpdateProcedures: + return procedureInfo.updateProcedures((UpdateProceduresPlan) physicalPlan); case UpdateProcedure: return procedureInfo.updateProcedure((UpdateProcedurePlan) physicalPlan); case DeleteProcedure: 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 0b5460486b29c..c6ec6a2fab602 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 @@ -537,7 +537,7 @@ private void executeProcedure(RootProcedureStack rootProcStack, Procedure p // state (e.g. AddRegionPeerProcedure parking at DO_ADD_REGION_PEER after waitTaskFinish() is // interrupted) would re-execute forever here and join() would hang. The persisted state lets // the next leader resume from where it stopped. - if (!isRunning() || !store.isRunning()) { + if (!isRunning() || !store.isRunning() || Thread.currentThread().isInterrupted()) { return; } @@ -603,13 +603,14 @@ private void updateStoreOnExecution( if (LOG.isDebugEnabled()) { LOG.debug(ProcedureMessages.STORED_CHILDREN, proc, Arrays.toString(subprocs)); } - try { - store.update(subprocs); - } catch (Exception e) { - // Do nothing since this step is idempotent. New CN leader can converge to the correct - // state when restore this procedure. - LOG.warn(ProcedureMessages.FAILED_TO_UPDATE_SUBPROCS_ON_EXECUTION, e); - } + // Persist the parent's next state and its children in one consensus entry before any child + // can run. Otherwise recovery can replay the parent's submission and create duplicate + // children. + Procedure[] proceduresToStore = Arrays.copyOf(subprocs, subprocs.length + 1); + proceduresToStore[subprocs.length] = proc; + RetryUtils.executeWithEndlessBackoffRetry( + () -> store.update(proceduresToStore), + ProcedureMessages.LOG_PERSIST_PARENT_AND_CHILD_PROCEDURES_54E2D0D2); } else { LOG.debug(ProcedureMessages.STORE_UPDATE, proc); if (proc.isFinished() && !proc.hasParent()) { diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ConfigProcedureStore.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ConfigProcedureStore.java index d9d3bf88cf87a..90fe25c204fa1 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ConfigProcedureStore.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ConfigProcedureStore.java @@ -19,23 +19,27 @@ package org.apache.iotdb.confignode.procedure.store; +import org.apache.iotdb.common.rpc.thrift.TSStatus; import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.commons.utils.FileUtils; import org.apache.iotdb.commons.utils.TestOnly; import org.apache.iotdb.confignode.consensus.request.write.procedure.DeleteProcedurePlan; import org.apache.iotdb.confignode.consensus.request.write.procedure.UpdateProcedurePlan; +import org.apache.iotdb.confignode.consensus.request.write.procedure.UpdateProceduresPlan; import org.apache.iotdb.confignode.i18n.ConfigNodeMessages; import org.apache.iotdb.confignode.manager.ConfigManager; import org.apache.iotdb.confignode.persistence.ProcedureInfo; import org.apache.iotdb.confignode.procedure.Procedure; import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv; import org.apache.iotdb.consensus.exception.ConsensusException; +import org.apache.iotdb.rpc.TSStatusCode; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.File; import java.io.IOException; +import java.util.Arrays; import java.util.List; import java.util.Objects; @@ -109,9 +113,18 @@ public void update(Procedure procedure) throws Exception } @Override - public void update(Procedure[] subprocs) throws Exception { - for (Procedure subproc : subprocs) { - update(subproc); + public void update(Procedure[] procedures) throws Exception { + for (Procedure procedure : procedures) { + Objects.requireNonNull( + ProcedureFactory.getProcedureType(procedure), + ConfigNodeMessages.EXCEPTION_PROCEDURE_TYPE_IS_NULL_93147BD3); + } + TSStatus status = + configManager + .getConsensusManager() + .write(new UpdateProceduresPlan(Arrays.asList(procedures))); + if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + throw new ConsensusException(status.toString()); } } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/IProcedureStore.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/IProcedureStore.java index 3dba6d29288d9..1b4ecb9934454 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/IProcedureStore.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/IProcedureStore.java @@ -40,7 +40,8 @@ public interface IProcedureStore { void update(Procedure procedure) throws Exception; - void update(Procedure[] subprocs) throws Exception; + /** Persist the entire batch atomically, including a parent and its newly created children. */ + void update(Procedure[] procedures) throws Exception; void delete(long procId) throws Exception; diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/UpdateProceduresPlanTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/UpdateProceduresPlanTest.java new file mode 100644 index 0000000000000..7b83c3b175d14 --- /dev/null +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/UpdateProceduresPlanTest.java @@ -0,0 +1,114 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.confignode.consensus.request; + +import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.confignode.consensus.request.write.procedure.UpdateProcedurePlan; +import org.apache.iotdb.confignode.consensus.request.write.procedure.UpdateProceduresPlan; +import org.apache.iotdb.confignode.manager.ConfigManager; +import org.apache.iotdb.confignode.manager.consensus.ConsensusManager; +import org.apache.iotdb.confignode.persistence.ProcedureInfo; +import org.apache.iotdb.confignode.procedure.Procedure; +import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv; +import org.apache.iotdb.confignode.procedure.impl.testonly.NeverFinishProcedure; +import org.apache.iotdb.confignode.procedure.store.ConfigProcedureStore; +import org.apache.iotdb.consensus.exception.ConsensusException; +import org.apache.iotdb.rpc.TSStatusCode; + +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.mockito.ArgumentCaptor; + +import java.io.File; +import java.util.Arrays; +import java.util.List; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +public class UpdateProceduresPlanTest { + @Rule public TemporaryFolder folder = new TemporaryFolder(); + + @Test + public void batchRoundTripAndSnapshotRetainAllProcedures() throws Exception { + List> procedures = + Arrays.asList( + new NeverFinishProcedure(7), new NeverFinishProcedure(8), new NeverFinishProcedure(4)); + UpdateProceduresPlan batch = new UpdateProceduresPlan(procedures); + UpdateProceduresPlan restored = + (UpdateProceduresPlan) ConfigPhysicalPlan.Factory.create(batch.serializeToByteBuffer()); + assertEquals(batch, restored); + ProcedureInfo info = new ProcedureInfo(null); + info.updateProcedures(restored); + File snapshot = folder.newFolder(); + assertTrue(info.processTakeSnapshot(snapshot)); + ProcedureInfo recovered = new ProcedureInfo(null); + recovered.processLoadSnapshot(snapshot); + assertEquals(info, recovered); + assertEquals(3, recovered.getProcedures().size()); + assertEquals(9, recovered.getNextProcId()); + // Existing single-procedure log entries retain their format and reader. + UpdateProcedurePlan single = new UpdateProcedurePlan(procedures.get(0)); + assertEquals(single, ConfigPhysicalPlan.Factory.create(single.serializeToByteBuffer())); + } + + @Test + public void storeWritesOneConsensusEntryForWholeBatch() throws Exception { + ConfigManager manager = mock(ConfigManager.class); + ConsensusManager consensus = mock(ConsensusManager.class); + when(manager.getConsensusManager()).thenReturn(consensus); + when(consensus.write(any())) + .thenReturn(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())); + ConfigProcedureStore store = new ConfigProcedureStore(manager, new ProcedureInfo(null)); + Procedure[] procedures = { + new NeverFinishProcedure(7), new NeverFinishProcedure(8), new NeverFinishProcedure(4) + }; + store.update(procedures); + ArgumentCaptor plan = ArgumentCaptor.forClass(ConfigPhysicalPlan.class); + verify(consensus).write(plan.capture()); + assertEquals( + Arrays.asList(procedures), ((UpdateProceduresPlan) plan.getValue()).getProcedures()); + } + + @Test + public void rejectedBatchIsNotReportedAsPersisted() throws Exception { + ConfigManager manager = mock(ConfigManager.class); + ConsensusManager consensus = mock(ConsensusManager.class); + when(manager.getConsensusManager()).thenReturn(consensus); + when(consensus.write(any())) + .thenReturn(new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode())); + ConfigProcedureStore store = new ConfigProcedureStore(manager, new ProcedureInfo(null)); + try { + store.update(new Procedure[] {new NeverFinishProcedure(7)}); + fail("Rejected consensus write must be retried by the executor"); + } catch (ConsensusException expected) { + assertTrue( + expected + .getMessage() + .contains(String.valueOf(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode()))); + } + } +} diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/ProcedureChildPersistenceTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/ProcedureChildPersistenceTest.java new file mode 100644 index 0000000000000..4f47eb872cfff --- /dev/null +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/ProcedureChildPersistenceTest.java @@ -0,0 +1,266 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.confignode.procedure; + +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.Test; + +import java.io.ByteArrayOutputStream; +import java.io.DataOutputStream; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +public class ProcedureChildPersistenceTest { + @Test + public void restartAfterPersistingOneChildDoesNotResubmit() throws Exception { + restartAfterPersistingChildren(1); + } + + @Test + public void restartAfterPersistingMultipleChildrenDoesNotResubmit() throws Exception { + restartAfterPersistingChildren(3); + } + + private void restartAfterPersistingChildren(int childCount) throws Exception { + ChildTestEnv env = new ChildTestEnv(childCount); + RecordingStore store = new RecordingStore(); + ProcedureExecutor executor = startExecutor(env, store); + CountDownLatch persisted = new CountDownLatch(1); + store.afterBatch = + () -> { + // Simulate a crash at the persistence boundary, before children are scheduled. + executor.stop(); + persisted.countDown(); + }; + long parentId; + try { + parentId = executor.submitProcedure(new ParentProcedure()); + assertTrue(persisted.await(10, TimeUnit.SECONDS)); + } finally { + executor.stop(); + executor.join(); + } + Procedure savedParent = store.records.get(parentId); + assertEquals(ProcedureState.WAITING, savedParent.getState()); + assertEquals(childCount + 1, store.records.size()); + assertEquals(0, env.childrenExecuted.get()); + + store.afterBatch = null; + ProcedureExecutor recovered = startExecutor(env, store); + try { + ProcedureTestUtil.waitForProcedure(recovered, parentId); + assertTrue(recovered.isFinished(parentId)); + assertEquals(1, env.submissions.get()); + assertEquals(childCount, env.childrenExecuted.get()); + assertEquals(1, env.parentCompleted.get()); + assertTrue( + recovered.getCompletedProcedures().stream() + .anyMatch(p -> p.getProcId() == parentId && p.isSuccess())); + } finally { + recovered.stop(); + recovered.join(); + store.stop(); + } + } + + @Test + public void failedBatchWriteIsRetriedBeforeChildrenRun() throws Exception { + ChildTestEnv env = new ChildTestEnv(3); + RecordingStore store = new RecordingStore(); + store.failFirstBatch = true; + ProcedureExecutor executor = startExecutor(env, store); + try { + long parentId = executor.submitProcedure(new ParentProcedure()); + ProcedureTestUtil.waitForProcedure(executor, parentId); + assertTrue(executor.isFinished(parentId)); + assertEquals(2, store.batchAttempts.get()); + assertEquals(1, env.submissions.get()); + assertEquals(3, env.childrenExecuted.get()); + assertEquals(1, env.parentCompleted.get()); + assertTrue( + executor.getCompletedProcedures().stream() + .anyMatch(p -> p.getProcId() == parentId && p.isSuccess())); + } finally { + executor.stop(); + executor.join(); + store.stop(); + } + } + + @Test + public void stoppingDuringFailedBatchWriteDoesNotStartChildren() throws Exception { + ChildTestEnv env = new ChildTestEnv(1); + RecordingStore store = new RecordingStore(); + store.failEveryBatch = true; + ProcedureExecutor executor = startExecutor(env, store); + try { + executor.submitProcedure(new ParentProcedure()); + assertTrue(store.batchAttempted.await(10, TimeUnit.SECONDS)); + } finally { + executor.stop(); + executor.join(); + store.stop(); + } + assertEquals(0, env.childrenExecuted.get()); + assertEquals(0, env.parentCompleted.get()); + assertEquals(1, store.records.size()); + assertEquals(ProcedureState.RUNNABLE, store.records.values().iterator().next().getState()); + } + + private ProcedureExecutor startExecutor(ChildTestEnv env, RecordingStore store) { + env.store = store; + store.start(); + ProcedureExecutor executor = new ProcedureExecutor<>(env, store); + executor.init(2); + executor.startWorkers(); + return executor; + } + + private static class ChildTestEnv extends TestProcEnv { + private final int childCount; + private final AtomicInteger submissions = new AtomicInteger(); + private final AtomicInteger childrenExecuted = new AtomicInteger(); + private final AtomicInteger parentCompleted = new AtomicInteger(); + private RecordingStore store; + + private ChildTestEnv(int childCount) { + this.childCount = childCount; + } + } + + private static class ParentProcedure extends StateMachineProcedure { + @Override + protected Flow executeFromState(ChildTestEnv env, Integer state) { + if (state == 0) { + env.submissions.incrementAndGet(); + for (int i = 0; i < env.childCount; i++) { + addChildProcedure(new ChildProcedure()); + } + setNextState(1); + return Flow.HAS_MORE_STATE; + } + assertEquals(env.childCount, env.childrenExecuted.get()); + env.parentCompleted.incrementAndGet(); + return Flow.NO_MORE_STATE; + } + + @Override + protected void rollbackState(ChildTestEnv env, Integer state) {} + + @Override + protected Integer getState(int stateId) { + return stateId; + } + + @Override + protected int getStateId(Integer state) { + return state; + } + + @Override + protected Integer getInitialState() { + return 0; + } + } + + private static class ChildProcedure extends Procedure { + @Override + protected Procedure[] execute(ChildTestEnv env) { + assertTrue(env.store.batchCommitted); + env.childrenExecuted.incrementAndGet(); + return null; + } + + @Override + protected void rollback(ChildTestEnv env) {} + } + + private static class RecordingStore extends NoopProcedureStore { + private final Map records = new HashMap<>(); + private final AtomicInteger batchAttempts = new AtomicInteger(); + private volatile boolean batchCommitted; + private boolean failFirstBatch; + private boolean failEveryBatch; + private final CountDownLatch batchAttempted = new CountDownLatch(1); + private Runnable afterBatch; + + @Override + public synchronized void update(Procedure procedure) { + records.put(procedure.getProcId(), copy(procedure)); + } + + @Override + public synchronized void update(Procedure[] procedures) { + int attempt = batchAttempts.incrementAndGet(); + batchAttempted.countDown(); + if (failEveryBatch || (attempt == 1 && failFirstBatch)) { + throw new IllegalStateException("Injected failure before committing the batch"); + } + for (Procedure procedure : procedures) { + update(procedure); + } + batchCommitted = true; + if (afterBatch != null) { + afterBatch.run(); + } + } + + @Override + public synchronized List load() { + List loaded = new ArrayList<>(); + records.values().forEach(procedure -> loaded.add(copy(procedure))); + return loaded; + } + + @Override + public synchronized void delete(long[] ids) { + for (long id : ids) { + records.remove(id); + } + } + + private Procedure copy(Procedure procedure) { + try { + ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + procedure.serialize(new DataOutputStream(bytes)); + Procedure restored = + procedure instanceof ParentProcedure ? new ParentProcedure() : new ChildProcedure(); + restored.deserialize(ByteBuffer.wrap(bytes.toByteArray())); + return restored; + } catch (IOException e) { + throw new AssertionError(e); + } + } + } +}