From 2186c619b5301ee2fb6c631b81c4ae73eba171a9 Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Wed, 4 Nov 2026 12:34:10 +0800 Subject: [PATCH 1/5] Fix pipe sink subtask silently failing to start after a rejected submission PipeSubtaskExecutor.start() marked the subtask as "submitting self" before calling submitSelf(). When the submission was rejected, e.g. because the worker pool had been shut down, the flag stayed set, so every later start() took the "already running" branch, logged nothing and reported success while the subtask never ran until the node restarted. The flag is now rolled back on failure and the rejection is logged and rethrown, so the failure reaches the pipe task agent and the config node instead of being masked as success. PipeAbstractSinkSubtask also clears its "isSubmitted" sign when submitting is disallowed, so a submission that was never dispatched to the worker pool can be submitted again. The IoTConsensusV2 subtask executor is shared by all the consensus pipes of a node. Once it had been shut down, getConsensusExecutor() kept returning the dead pool, so every consensus pipe registered afterwards could never start its sink subtask. A shut-down executor is now recreated and installed back into IoTV2GlobalComponentContainer. --- .../iotdb/db/i18n/DataNodePipeMessages.java | 2 + .../iotdb/db/i18n/DataNodePipeMessages.java | 2 + .../execution/PipeSubtaskExecutorManager.java | 28 ++++++++- .../task/PipeSinkSubtaskExecutorTest.java | 24 ++++++++ .../agent/task/PipeSubtaskExecutorTest.java | 22 +++++++ .../PipeSubtaskExecutorManagerTest.java | 59 +++++++++++++++++++ .../iotdb/commons/i18n/PipeMessages.java | 2 + .../iotdb/commons/i18n/PipeMessages.java | 2 + .../task/execution/PipeSubtaskExecutor.java | 18 +++++- .../task/subtask/PipeAbstractSinkSubtask.java | 9 +++ 10 files changed, 163 insertions(+), 5 deletions(-) create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/execution/PipeSubtaskExecutorManagerTest.java diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java index f56589e52c0db..8e7243d2c15aa 100644 --- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java +++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java @@ -358,6 +358,8 @@ public final class DataNodePipeMessages { "Successfully persisted all pipe's info to configNode."; public static final String THE_EXECUTOR_AND_HAS_BEEN_SUCCESSFULLY_SHUTDOWN = "The executor {} and {} has been successfully shutdown."; + public static final String LOG_THE_SHARED_CONSENSUS_EXECUTOR_HAS_BEEN_SHUT_DOWN_RECREATING_IT_C14080E3 = + "The shared consensus executor has been shut down. Recreating it."; // ===================== EVENT ===================== diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java index 1b8b488f30f17..ab67d7ab46ed7 100644 --- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java +++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java @@ -340,6 +340,8 @@ public final class DataNodePipeMessages { "成功将所有 Pipe 信息持久化到 ConfigNode。"; public static final String THE_EXECUTOR_AND_HAS_BEEN_SUCCESSFULLY_SHUTDOWN = "执行器 {} 和 {} 已成功关闭。"; + public static final String LOG_THE_SHARED_CONSENSUS_EXECUTOR_HAS_BEEN_SHUT_DOWN_RECREATING_IT_C14080E3 = + "共享的共识执行器已被关闭，正在重新创建该执行器。"; // ===================== EVENT ===================== diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/execution/PipeSubtaskExecutorManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/execution/PipeSubtaskExecutorManager.java index f83c23871f516..36a8602dea15c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/execution/PipeSubtaskExecutorManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/execution/PipeSubtaskExecutorManager.java @@ -20,10 +20,15 @@ package org.apache.iotdb.db.pipe.agent.task.execution; import org.apache.iotdb.commons.consensus.iotv2.container.IoTV2GlobalComponentContainer; +import org.apache.iotdb.commons.pipe.agent.task.execution.PipeSubtaskExecutor; import org.apache.iotdb.commons.subscription.config.SubscriptionConfig; +import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.consensus.IoTConsensusV2SubtaskExecutor; import org.apache.iotdb.db.subscription.task.execution.SubscriptionSubtaskExecutor; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import java.util.function.Supplier; /** @@ -31,6 +36,9 @@ * PipeTaskScheduler. It is a singleton class. */ public class PipeSubtaskExecutorManager { + + private static final Logger LOGGER = LoggerFactory.getLogger(PipeSubtaskExecutorManager.class); + private final PipeProcessorSubtaskExecutor processorExecutor; private final Supplier connectorExecutorSupplier; private volatile SubscriptionSubtaskExecutor subscriptionExecutor; @@ -43,9 +51,23 @@ public Supplier getConnectorExecutorSupplier() { return connectorExecutorSupplier; } - public IoTConsensusV2SubtaskExecutor getConsensusExecutor() { - return (IoTConsensusV2SubtaskExecutor) - IoTV2GlobalComponentContainer.getInstance().getConsensusExecutor(); + public synchronized IoTConsensusV2SubtaskExecutor getConsensusExecutor() { + final IoTV2GlobalComponentContainer container = IoTV2GlobalComponentContainer.getInstance(); + final PipeSubtaskExecutor consensusExecutor = container.getConsensusExecutor(); + if (consensusExecutor instanceof IoTConsensusV2SubtaskExecutor + && !consensusExecutor.isShutdown()) { + return (IoTConsensusV2SubtaskExecutor) consensusExecutor; + } + + // The consensus executor is shared by all the consensus pipes of the node. If it has been shut + // down unexpectedly, recreate it, otherwise all the consensus pipes created or restarted + // afterwards would never be able to submit their sink subtasks. + LOGGER.warn( + DataNodePipeMessages + .LOG_THE_SHARED_CONSENSUS_EXECUTOR_HAS_BEEN_SHUT_DOWN_RECREATING_IT_C14080E3); + final IoTConsensusV2SubtaskExecutor newConsensusExecutor = new IoTConsensusV2SubtaskExecutor(); + container.setConsensusExecutor(newConsensusExecutor); + return newConsensusExecutor; } public SubscriptionSubtaskExecutor getSubscriptionExecutor() { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeSinkSubtaskExecutorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeSinkSubtaskExecutorTest.java index 9cd54b425cf74..784fe39c4ee24 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeSinkSubtaskExecutorTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeSinkSubtaskExecutorTest.java @@ -24,9 +24,13 @@ import org.apache.iotdb.db.pipe.agent.task.subtask.sink.PipeSinkSubtask; import org.apache.iotdb.pipe.api.PipeConnector; +import org.junit.Assert; import org.junit.Before; +import org.junit.Test; import org.mockito.Mockito; +import java.util.concurrent.RejectedExecutionException; + import static org.mockito.Mockito.mock; public class PipeSinkSubtaskExecutorTest extends PipeSubtaskExecutorTest { @@ -46,4 +50,24 @@ public void setUp() throws Exception { mock(UnboundedBlockingPendingQueue.class), mock(PipeConnector.class))); } + + @Test + public void testStartAfterShutdown() { + executor.register(subtask); + executor.shutdown(); + Assert.assertTrue(executor.isShutdown()); + + // The submission is rejected by the shut-down executor, and the rejection must be reported + // instead of being swallowed as "started successfully"... + for (int i = 0; i < 2; ++i) { + try { + executor.start(subtask.getTaskID()); + Assert.fail("The start of a subtask should fail if the executor is shut down"); + } catch (final RejectedExecutionException e) { + // expected + } + // ... and the failed submission must not leave the subtask in the "submitting" state. + Assert.assertFalse(subtask.isSubmittingSelf()); + } + } } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeSubtaskExecutorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeSubtaskExecutorTest.java index 892c61536e17d..88838cbd75d98 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeSubtaskExecutorTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeSubtaskExecutorTest.java @@ -25,6 +25,9 @@ import org.junit.After; import org.junit.Assert; import org.junit.Test; +import org.mockito.Mockito; + +import java.util.concurrent.RejectedExecutionException; import static org.mockito.Mockito.atLeast; import static org.mockito.Mockito.times; @@ -115,6 +118,25 @@ public void testStop() { Assert.assertFalse(subtask.isSubmittingSelf()); } + @Test + public void testStartWithRejectedSubmission() { + executor.register(subtask); + Mockito.doThrow(new RejectedExecutionException()).when(subtask).submitSelf(); + + try { + executor.start(subtask.getTaskID()); + Assert.fail("The start of a subtask should fail if the submission is rejected"); + } catch (final RejectedExecutionException e) { + // expected + } + + // The failed submission must be rolled back, otherwise the following start calls would take + // the "already running" branch and report a false success. + verify(subtask).allowSubmittingSelf(); + verify(subtask).disallowSubmittingSelf(); + Assert.assertFalse(executor.isShutdown()); + } + @Test public void testDeregister() { // test unregister a subtask which is not in the map diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/execution/PipeSubtaskExecutorManagerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/execution/PipeSubtaskExecutorManagerTest.java new file mode 100644 index 0000000000000..59eee60128e66 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/execution/PipeSubtaskExecutorManagerTest.java @@ -0,0 +1,59 @@ +/* + * 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.db.pipe.agent.task.execution; + +import org.apache.iotdb.db.pipe.consensus.IoTConsensusV2SubtaskExecutor; + +import org.junit.Assert; +import org.junit.Test; + +public class PipeSubtaskExecutorManagerTest { + + @Test + public void testGetConsensusExecutor() { + final PipeSubtaskExecutorManager manager = PipeSubtaskExecutorManager.getInstance(); + + final IoTConsensusV2SubtaskExecutor consensusExecutor = manager.getConsensusExecutor(); + Assert.assertFalse(consensusExecutor.isShutdown()); + // The consensus executor is shared by all the consensus pipes, so it must be a singleton as + // long as it is alive. + Assert.assertSame(consensusExecutor, manager.getConsensusExecutor()); + + consensusExecutor.shutdown(); + Assert.assertTrue(consensusExecutor.isShutdown()); + } + + @Test + public void testGetConsensusExecutorRecoversFromShutdown() { + final PipeSubtaskExecutorManager manager = PipeSubtaskExecutorManager.getInstance(); + + final IoTConsensusV2SubtaskExecutor consensusExecutor = manager.getConsensusExecutor(); + consensusExecutor.shutdown(); + Assert.assertTrue(consensusExecutor.isShutdown()); + + // A shut-down consensus executor must not be handed to the following consensus pipes, + // otherwise none of their sink subtasks could ever be started. + final IoTConsensusV2SubtaskExecutor recoveredExecutor = manager.getConsensusExecutor(); + Assert.assertNotSame(consensusExecutor, recoveredExecutor); + Assert.assertFalse(recoveredExecutor.isShutdown()); + + recoveredExecutor.shutdown(); + } +} diff --git a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java index a14719422788f..52795a1ed7217 100644 --- a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java @@ -390,6 +390,8 @@ public final class PipeMessages { public static final String SUBTASK_STARTED = "The subtask {} is started to submit self."; public static final String SUBTASK_CLOSED = "The subtask {} is closed successfully."; public static final String SUBTASK_CLOSE_FAILED = "Failed to close the subtask {}."; + public static final String LOG_FAILED_TO_START_THE_SUBTASK_DD3FC1BA = + "Failed to start the subtask {}."; // ===================== PipeReportableSubtask ===================== diff --git a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java index 47ddd88162ddc..d95d7b8b95a77 100644 --- a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java @@ -374,6 +374,8 @@ public final class PipeMessages { public static final String SUBTASK_STARTED = "子任务 {} 已开始自提交。"; public static final String SUBTASK_CLOSED = "子任务 {} 已成功关闭。"; public static final String SUBTASK_CLOSE_FAILED = "关闭子任务 {} 失败。"; + public static final String LOG_FAILED_TO_START_THE_SUBTASK_DD3FC1BA = + "启动子任务 {} 失败。"; // ===================== PipeReportableSubtask ===================== diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/execution/PipeSubtaskExecutor.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/execution/PipeSubtaskExecutor.java index 7fe169f7a38fe..afa3b33b0000f 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/execution/PipeSubtaskExecutor.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/execution/PipeSubtaskExecutor.java @@ -144,8 +144,22 @@ public final synchronized void start(final String subTaskID) { LOGGER.debug(PipeMessages.SUBTASK_ALREADY_RUNNING, subtask.getDisplayTaskID()); } } else { - subtask.allowSubmittingSelf(); - subtask.submitSelf(); + try { + subtask.allowSubmittingSelf(); + subtask.submitSelf(); + } catch (final RuntimeException e) { + // The submission may be rejected, e.g., because the underlying thread pool has been shut + // down. The "submitting self" state must be reverted, otherwise this and all the following + // start() calls would take the "already running" branch above and report a false success, + // leaving the subtask never executed until the node restarts. + subtask.disallowSubmittingSelf(); + LOGGER.error( + PipeMessages.LOG_FAILED_TO_START_THE_SUBTASK_DD3FC1BA, subtask.getDisplayTaskID(), e); + throw e; + } catch (final Error e) { + subtask.disallowSubmittingSelf(); + throw e; + } ++runningSubtaskNumber; LOGGER.info(PipeMessages.SUBTASK_STARTED, subtask.getDisplayTaskID()); } diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java index 4d7f16779e3d6..c607722c7f4ac 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java @@ -83,6 +83,15 @@ public void bindExecutors( this.subtaskScheduler = subtaskScheduler; } + @Override + protected synchronized void onDisallowSubmittingSelf() { + // The subtask is not submitting itself any more. The "isSubmitted" sign is cleared so that a + // later allowSubmittingSelf() + submitSelf() can really submit the subtask again, even if the + // previous submission was never dispatched to the worker pool, e.g., because the submission was + // rejected or scheduled on an executor that has been shut down. + isSubmitted = false; + } + @Override public void onSuccess(final Boolean hasAtLeastOneEventProcessed) { preScheduleLowPriorityTask(100); From 83dc587b8dfbc3e1aa99051e5f9463aa6cac3164 Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Wed, 4 Nov 2026 12:34:10 +0800 Subject: [PATCH 2/5] Retry pipe tasks whose start failed instead of leaving them dead startPipeTask() let an exception of PipeTask.start() escape without rolling back the task it had just created. The task meta was already registered, so the following meta sync found an existing task meta with the same leader and never called start() again: the pipe stayed "RUNNING" in the metadata while running nothing until the node restarted. The failed task is now dropped together with its task meta, so that the next meta sync recreates and starts the task again, which gives the task another chance to start once the underlying problem is gone. The failure itself is logged at error level instead of being carried away by the meta sync loop. --- .../agent/task/PipeDataNodeTaskAgentTest.java | 40 +++++++++++++++++++ .../iotdb/commons/i18n/PipeMessages.java | 7 ++++ .../iotdb/commons/i18n/PipeMessages.java | 6 +++ .../pipe/agent/task/PipeTaskAgent.java | 37 ++++++++++++++++- 4 files changed, 88 insertions(+), 2 deletions(-) diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgentTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgentTest.java index fe1bfbe4aabe8..4e5ccca3a5009 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgentTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgentTest.java @@ -25,6 +25,7 @@ import org.apache.iotdb.commons.consensus.index.impl.SimpleProgressIndex; import org.apache.iotdb.commons.exception.pipe.PipeRuntimeCriticalException; import org.apache.iotdb.commons.exception.pipe.PipeRuntimeSinkCriticalException; +import org.apache.iotdb.commons.pipe.agent.task.PipeTask; import org.apache.iotdb.commons.pipe.agent.task.PipeTaskAgent; import org.apache.iotdb.commons.pipe.agent.task.PipeTaskManager; import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta; @@ -41,9 +42,11 @@ import org.awaitility.Awaitility; import org.junit.Assert; import org.junit.Test; +import org.mockito.Mockito; import java.io.IOException; import java.lang.reflect.Field; +import java.lang.reflect.Method; import java.util.Arrays; import java.util.Collections; import java.util.HashMap; @@ -576,6 +579,43 @@ public void testCarryOverRequiresStableLeaderAndLocalTask() { .getProgressIndex()); } + @Test + public void testFailedTaskStartIsRolledBackForRetry() throws Exception { + final PipeDataNodeTaskAgent agent = newActiveTaskAgent(); + final PipeMetaKeeper keeper = getField(agent, "pipeMetaKeeper"); + final PipeTaskManager manager = getField(agent, "pipeTaskManager"); + + final int localNodeId = IoTDBDescriptor.getInstance().getConfig().getDataNodeId(); + final PipeTaskMeta taskMeta = new PipeTaskMeta(MinimumProgressIndex.INSTANCE, localNodeId); + final ConcurrentMap taskMetaMap = new ConcurrentHashMap<>(); + taskMetaMap.put(REGION_ID, taskMeta); + final PipeMeta pipeMeta = + new PipeMeta( + new PipeStaticMeta( + "failedStartPipe", 1L, new HashMap<>(), new HashMap<>(), new HashMap<>()), + new PipeRuntimeMeta(taskMetaMap)); + pipeMeta.getRuntimeMeta().getStatus().set(PipeStatus.RUNNING); + keeper.addPipeMeta(pipeMeta); + + // A task which has been created but can not be started, e.g., the sink of the task has been + // rejected by a shut-down thread pool. + final PipeTask failedTask = Mockito.mock(PipeTask.class); + Mockito.doThrow(new PipeException("failed to start the sink")).when(failedTask).start(); + manager.addPipeTask(pipeMeta.getStaticMeta(), REGION_ID, failedTask); + + final Method startPipeTask = + PipeTaskAgent.class.getDeclaredMethod("startPipeTask", int.class, PipeStaticMeta.class); + startPipeTask.setAccessible(true); + startPipeTask.invoke(agent, REGION_ID, pipeMeta.getStaticMeta()); + + // The half-started task is dropped and its task meta is removed, so that the following meta + // sync creates and starts the task again instead of keeping it as "RUNNING but never runs". + Mockito.verify(failedTask).drop(); + Assert.assertNull(manager.getPipeTask(pipeMeta.getStaticMeta(), REGION_ID)); + Assert.assertNull(pipeMeta.getRuntimeMeta().getConsensusGroupId2TaskMetaMap().get(REGION_ID)); + Assert.assertEquals(PipeStatus.RUNNING, pipeMeta.getRuntimeMeta().getStatus().get()); + } + private PipeStaticMeta createStaticMeta(final long creationTime, final boolean historyEnabled) { final Map sourceAttributes = new HashMap<>(); sourceAttributes.put( diff --git a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java index 52795a1ed7217..92f7d242e1482 100644 --- a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java @@ -364,6 +364,13 @@ public final class PipeMessages { public static final String PIPE_WAS_STOPPED_CRITICAL_EXCEPTION = "Pipe %s (creation time = %s) was stopped because of critical exception " + "(occurred time %s)."; + public static final String + LOG_FAILED_TO_START_THE_TASK_OF_CONSENSUS_GROUP_ARG_FOR_PIPE_ARG_THE_TASK_IS_DROPPED_AND_WILL_BE_RECREATED_BY_THE_FOLLOWING_META_SYNC_7594577C = + "Failed to start the task of consensus group {} for pipe {}. The task is dropped and " + + "will be recreated by the following meta sync."; + public static final String + LOG_FAILED_TO_DROP_THE_TASK_OF_CONSENSUS_GROUP_ARG_FOR_PIPE_ARG_AFTER_IT_FAILED_TO_START_394297EC = + "Failed to drop the task of consensus group {} for pipe {} after it failed to start."; // ===================== PipeEventCommitManager ===================== diff --git a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java index d95d7b8b95a77..985977ece5750 100644 --- a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java @@ -349,6 +349,12 @@ public final class PipeMessages { "Pipe %s（创建时间 = %s）将因 connector %s 中的严重异常（发生时间 %s）而停止。"; public static final String PIPE_WAS_STOPPED_CRITICAL_EXCEPTION = "Pipe %s（创建时间 = %s）已因严重异常（发生时间 %s）而停止。"; + public static final String + LOG_FAILED_TO_START_THE_TASK_OF_CONSENSUS_GROUP_ARG_FOR_PIPE_ARG_THE_TASK_IS_DROPPED_AND_WILL_BE_RECREATED_BY_THE_FOLLOWING_META_SYNC_7594577C = + "启动共识组 {} 的 Pipe {} 任务失败。该任务已被丢弃，将由下一次元数据同步重新创建。"; + public static final String + LOG_FAILED_TO_DROP_THE_TASK_OF_CONSENSUS_GROUP_ARG_FOR_PIPE_ARG_AFTER_IT_FAILED_TO_START_394297EC = + "启动失败后，丢弃共识组 {} 的 Pipe {} 任务时出错。"; // ===================== PipeEventCommitManager ===================== diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java index 8352d2c6be1fb..e05cb78fb3cf5 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java @@ -977,9 +977,42 @@ private void dropPipeTask(final int consensusGroupId, final PipeStaticMeta pipeS private void startPipeTask(final int consensusGroupId, final PipeStaticMeta pipeStaticMeta) { final PipeTask pipeTask = pipeTaskManager.getPipeTask(pipeStaticMeta, consensusGroupId); - if (pipeTask != null) { + if (pipeTask == null) { + return; + } + + try { pipeTask.start(); - thawRate(pipeStaticMeta.getPipeName(), pipeStaticMeta.getCreationTime()); + } catch (final Exception e) { + LOGGER.error( + PipeMessages + .LOG_FAILED_TO_START_THE_TASK_OF_CONSENSUS_GROUP_ARG_FOR_PIPE_ARG_THE_TASK_IS_DROPPED_AND_WILL_BE_RECREATED_BY_THE_FOLLOWING_META_SYNC_7594577C, + consensusGroupId, + pipeStaticMeta.getPipeName(), + e); + + // The task is not running now. Dropping it instead of keeping it as is, because a task that + // has already been created is never started again by the following meta sync, which would + // leave the pipe alive in the metadata but dead in fact until the node restarts. After the + // task meta is removed here, the following meta sync sees no task meta of this consensus + // group and creates the task again, which gives the task another chance to start. + rollbackPipeTask(consensusGroupId, pipeStaticMeta); + return; + } + + thawRate(pipeStaticMeta.getPipeName(), pipeStaticMeta.getCreationTime()); + } + + private void rollbackPipeTask(final int consensusGroupId, final PipeStaticMeta pipeStaticMeta) { + try { + dropPipeTask(consensusGroupId, pipeStaticMeta); + } catch (final Exception e) { + LOGGER.warn( + PipeMessages + .LOG_FAILED_TO_DROP_THE_TASK_OF_CONSENSUS_GROUP_ARG_FOR_PIPE_ARG_AFTER_IT_FAILED_TO_START_394297EC, + consensusGroupId, + pipeStaticMeta.getPipeName(), + e); } } From f67c5bd8a89e3d01e20176fe475686aa698d7873 Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Wed, 4 Nov 2026 13:14:52 +0800 Subject: [PATCH 3/5] Report the rejection of a pipe sink subtask submission instead of stopping silently A sink subtask submits itself again after the end of each round. If the submission is rejected, e.g. because the worker pool is being shut down while the subtask is finishing a round, the rejection escaped from the callback thread where it was swallowed, and, for a delayed submission, the "submitted" sign had already been set before scheduling the delayed task, so it was never cleared. In both cases the subtask was never submitted again and stopped transferring events without leaving any trace in the log. The sign is now cleared and the rejection is reported whenever the submission or the delayed scheduling fails, so the failure is visible and a following start of the subtask can submit it again. --- .../task/PipeSinkSubtaskExecutorTest.java | 68 +++++++++++++++++++ .../iotdb/commons/i18n/PipeMessages.java | 3 + .../iotdb/commons/i18n/PipeMessages.java | 3 + .../task/subtask/PipeAbstractSinkSubtask.java | 35 +++++++--- 4 files changed, 99 insertions(+), 10 deletions(-) diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeSinkSubtaskExecutorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeSinkSubtaskExecutorTest.java index 784fe39c4ee24..b8c63ece8a518 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeSinkSubtaskExecutorTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeSinkSubtaskExecutorTest.java @@ -20,6 +20,8 @@ package org.apache.iotdb.db.pipe.agent.task; import org.apache.iotdb.commons.pipe.agent.task.connection.UnboundedBlockingPendingQueue; +import org.apache.iotdb.commons.pipe.agent.task.subtask.PipeSubtask; +import org.apache.iotdb.commons.pipe.sink.protocol.PipeSinkWithSchedulingDelay; import org.apache.iotdb.db.pipe.agent.task.execution.PipeSinkSubtaskExecutor; import org.apache.iotdb.db.pipe.agent.task.subtask.sink.PipeSinkSubtask; import org.apache.iotdb.pipe.api.PipeConnector; @@ -29,7 +31,9 @@ import org.junit.Test; import org.mockito.Mockito; +import java.lang.reflect.Field; import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.atomic.AtomicBoolean; import static org.mockito.Mockito.mock; @@ -70,4 +74,68 @@ public void testStartAfterShutdown() { Assert.assertFalse(subtask.isSubmittingSelf()); } } + + @Test + public void testSubmitSelfAfterShutdown() throws Exception { + executor.register(subtask); + executor.shutdown(); + allowSubmittingSelfAgain(subtask); + + // The subtask submits itself again after each round, so a rejected submission must be reported + // instead of being swallowed. Otherwise the subtask would stop transferring events silently. + for (int i = 0; i < 2; ++i) { + try { + subtask.submitSelf(); + Assert.fail("The submission of a subtask should fail if the executor is shut down"); + } catch (final RejectedExecutionException e) { + // expected + } + } + } + + @Test + public void testDelayedSubmitSelfAfterShutdown() throws Exception { + final SinkWithSchedulingDelay sink = mock(SinkWithSchedulingDelay.class); + Mockito.when(sink.consumeSchedulingDelayMs()).thenReturn(1000L); + + final PipeSubtask delayedSubtask = + Mockito.spy( + new PipeSinkSubtask( + "PipeConnectorSubtaskExecutorTest", + System.currentTimeMillis(), + "TestAttributeSortedString", + "TestAttributeSortedString", + 0, + mock(UnboundedBlockingPendingQueue.class), + sink)); + + executor.register(delayedSubtask); + executor.shutdown(); + allowSubmittingSelfAgain(delayedSubtask); + + // A delayed submission marks the subtask as submitted before scheduling it. If the scheduling + // is rejected, the sign must be cleared as well, otherwise every following submission would be + // skipped and the subtask would never be submitted again. + for (int i = 0; i < 2; ++i) { + try { + delayedSubtask.submitSelf(); + Assert.fail("The submission of a subtask should fail if the executor is shut down"); + } catch (final RejectedExecutionException e) { + // expected + } + } + } + + /** + * Simulates the race where a subtask is submitting itself while its executor is being shut down: + * the executor can not submit it any more, but the subtask has not been told to stop yet. + */ + private void allowSubmittingSelfAgain(final PipeSubtask subtask) throws Exception { + final Field shouldStopSubmittingSelfField = + PipeSubtask.class.getDeclaredField("shouldStopSubmittingSelf"); + shouldStopSubmittingSelfField.setAccessible(true); + ((AtomicBoolean) shouldStopSubmittingSelfField.get(subtask)).set(false); + } + + private interface SinkWithSchedulingDelay extends PipeConnector, PipeSinkWithSchedulingDelay {} } diff --git a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java index 92f7d242e1482..e9f0dda0fba16 100644 --- a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java @@ -396,6 +396,9 @@ public final class PipeMessages { public static final String SUBTASK_ALREADY_RUNNING = "The subtask {} is already running."; public static final String SUBTASK_STARTED = "The subtask {} is started to submit self."; public static final String SUBTASK_CLOSED = "The subtask {} is closed successfully."; + public static final String + LOG_FAILED_TO_SUBMIT_THE_SUBTASK_ARG_BECAUSE_THE_EXECUTOR_HAS_BEEN_SHUT_DOWN_198335BE = + "Failed to submit the subtask {} because the executor has been shut down."; public static final String SUBTASK_CLOSE_FAILED = "Failed to close the subtask {}."; public static final String LOG_FAILED_TO_START_THE_SUBTASK_DD3FC1BA = "Failed to start the subtask {}."; diff --git a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java index 985977ece5750..8e09d70739e14 100644 --- a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java @@ -379,6 +379,9 @@ public final class PipeMessages { public static final String SUBTASK_ALREADY_RUNNING = "子任务 {} 已在运行。"; public static final String SUBTASK_STARTED = "子任务 {} 已开始自提交。"; public static final String SUBTASK_CLOSED = "子任务 {} 已成功关闭。"; + public static final String + LOG_FAILED_TO_SUBMIT_THE_SUBTASK_ARG_BECAUSE_THE_EXECUTOR_HAS_BEEN_SHUT_DOWN_198335BE = + "执行器已关闭，提交子任务 {} 失败。"; public static final String SUBTASK_CLOSE_FAILED = "关闭子任务 {} 失败。"; public static final String LOG_FAILED_TO_START_THE_SUBTASK_DD3FC1BA = "启动子任务 {} 失败。"; diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java index c607722c7f4ac..95863a79da9c3 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java @@ -42,6 +42,7 @@ import org.slf4j.LoggerFactory; import java.util.concurrent.ExecutorService; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.TimeUnit; public abstract class PipeAbstractSinkSubtask extends PipeReportableSubtask { @@ -288,17 +289,31 @@ public synchronized void submitSelf() { return; } - final long schedulingDelayInMs = getNextSchedulingDelayInMs(); - if (schedulingDelayInMs > 0) { - isSubmitted = true; - subtaskWorkerScheduledExecutor.schedule( - // Keep the isSubmitted placeholder set before the delayed submission to avoid duplicate - // schedules, so the delayed task should not mark it again. - () -> submitSelfToWorker(false), schedulingDelayInMs, TimeUnit.MILLISECONDS); - return; - } + try { + final long schedulingDelayInMs = getNextSchedulingDelayInMs(); + if (schedulingDelayInMs > 0) { + isSubmitted = true; + subtaskWorkerScheduledExecutor.schedule( + // Keep the isSubmitted placeholder set before the delayed submission to avoid duplicate + // schedules, so the delayed task should not mark it again. + () -> submitSelfToWorker(false), schedulingDelayInMs, TimeUnit.MILLISECONDS); + return; + } - submitSelfToWorker(true); + submitSelfToWorker(true); + } catch (final RejectedExecutionException e) { + // The worker pool or the scheduler has been shut down, e.g., because the pipe has been + // dropped. The "isSubmitted" sign must be cleared, otherwise the subtask would never be + // submitted again and would silently stop transferring events. The rejection is reported + // instead of being swallowed by the callback executor. + isSubmitted = false; + LOGGER.warn( + PipeMessages + .LOG_FAILED_TO_SUBMIT_THE_SUBTASK_ARG_BECAUSE_THE_EXECUTOR_HAS_BEEN_SHUT_DOWN_198335BE, + getDisplayTaskID(), + e); + throw e; + } } @Override From 0e04990522bd30ee851b21326659706ecebeb20e Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Wed, 4 Nov 2026 14:47:36 +0800 Subject: [PATCH 4/5] Fix pipe subtask single-flight violation when stopped and restarted PipeAbstractSinkSubtask#onDisallowSubmittingSelf cleared isSubmitted while a submitted call could still be running. A following start() then submitted the same subtask again, so two workers could execute one sink subtask concurrently, which PipeSubtask does not support. Keep isSubmitted as the single-flight flag: a call in flight is allowed to finish and re-submit itself once start() has cleared the stop flag. Instead, make PipeSubtaskExecutor#start fail fast when the executor has been shut down, before touching any subtask state. A shut-down executor can never run the subtask, so start() must not report a false success. This also covers a delayed submission that is still pending when the executor is shut down. Add regression tests for stopping and restarting a subtask while its first call is blocked, and for starting a subtask with a pending delayed submission after shutdown. Recreate the shared consensus executor after the manager tests so they do not leave it shut down for other tests. --- .../task/PipeSinkSubtaskExecutorTest.java | 91 +++++++++++++++++++ .../PipeSubtaskExecutorManagerTest.java | 8 ++ .../iotdb/commons/i18n/PipeMessages.java | 3 + .../iotdb/commons/i18n/PipeMessages.java | 3 + .../task/execution/PipeSubtaskExecutor.java | 12 +++ .../task/subtask/PipeAbstractSinkSubtask.java | 9 -- 6 files changed, 117 insertions(+), 9 deletions(-) diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeSinkSubtaskExecutorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeSinkSubtaskExecutorTest.java index b8c63ece8a518..7a97f029fb7ad 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeSinkSubtaskExecutorTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeSinkSubtaskExecutorTest.java @@ -32,8 +32,11 @@ import org.mockito.Mockito; import java.lang.reflect.Field; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import static org.mockito.Mockito.mock; @@ -126,6 +129,94 @@ public void testDelayedSubmitSelfAfterShutdown() throws Exception { } } + @Test + public void testStartWhileThePreviousCallIsStillRunning() throws Exception { + final AtomicInteger concurrentCallCount = new AtomicInteger(0); + final AtomicInteger maxConcurrentCallCount = new AtomicInteger(0); + final CountDownLatch firstCallEntered = new CountDownLatch(1); + final CountDownLatch firstCallReleased = new CountDownLatch(1); + final CountDownLatch secondCallEntered = new CountDownLatch(1); + + final PipeSubtask blockedSubtask = + new PipeSinkSubtask( + "PipeConnectorSubtaskExecutorTest", + System.currentTimeMillis(), + "TestAttributeSortedString", + "TestAttributeSortedString", + 0, + mock(UnboundedBlockingPendingQueue.class), + mock(PipeConnector.class)) { + @Override + protected boolean executeOnce() { + final int current = concurrentCallCount.incrementAndGet(); + maxConcurrentCallCount.accumulateAndGet(current, Math::max); + if (current == 1) { + firstCallEntered.countDown(); + try { + firstCallReleased.await(5, TimeUnit.SECONDS); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + } + } else { + secondCallEntered.countDown(); + } + concurrentCallCount.decrementAndGet(); + return false; + } + }; + + executor.register(blockedSubtask); + executor.start(blockedSubtask.getTaskID()); + Assert.assertTrue(firstCallEntered.await(5, TimeUnit.SECONDS)); + + // Stop and start while the first call is still running. The subtask must not be submitted + // again, because a PipeSubtask is not designed to run in parallel with itself. + executor.stop(blockedSubtask.getTaskID()); + executor.start(blockedSubtask.getTaskID()); + + try { + Assert.assertFalse( + "The subtask must not be executed in parallel with itself", + secondCallEntered.await(1, TimeUnit.SECONDS)); + Assert.assertEquals(1, maxConcurrentCallCount.get()); + } finally { + executor.stop(blockedSubtask.getTaskID()); + firstCallReleased.countDown(); + } + } + + @Test + public void testStartAfterShutdownWithPendingDelayedSubmission() { + final SinkWithSchedulingDelay sink = mock(SinkWithSchedulingDelay.class); + Mockito.when(sink.consumeSchedulingDelayMs()).thenReturn(60_000L); + + final PipeSubtask delayedSubtask = + new PipeSinkSubtask( + "PipeConnectorSubtaskExecutorTest", + System.currentTimeMillis(), + "TestAttributeSortedString", + "TestAttributeSortedString", + 0, + mock(UnboundedBlockingPendingQueue.class), + sink); + + executor.register(delayedSubtask); + + // The delayed submission is accepted and marks the subtask as submitted before it is + // dispatched. + executor.start(delayedSubtask.getTaskID()); + + // The delayed submission will never be dispatched to the shut-down worker pool, so a following + // start() must not take the "already submitted" branch and report a false success. + executor.shutdown(); + try { + executor.start(delayedSubtask.getTaskID()); + Assert.fail("The start of a subtask should fail if the executor is shut down"); + } catch (final RejectedExecutionException e) { + // expected + } + } + /** * Simulates the race where a subtask is submitting itself while its executor is being shut down: * the executor can not submit it any more, but the subtask has not been told to stop yet. diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/execution/PipeSubtaskExecutorManagerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/execution/PipeSubtaskExecutorManagerTest.java index 59eee60128e66..21267f305666b 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/execution/PipeSubtaskExecutorManagerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/execution/PipeSubtaskExecutorManagerTest.java @@ -21,11 +21,19 @@ import org.apache.iotdb.db.pipe.consensus.IoTConsensusV2SubtaskExecutor; +import org.junit.After; import org.junit.Assert; import org.junit.Test; public class PipeSubtaskExecutorManagerTest { + @After + public void tearDown() { + // Do not leave the node-wide shared consensus executor shut down for the other tests in the + // same JVM. getConsensusExecutor() recreates it if it has been shut down. + PipeSubtaskExecutorManager.getInstance().getConsensusExecutor(); + } + @Test public void testGetConsensusExecutor() { final PipeSubtaskExecutorManager manager = PipeSubtaskExecutorManager.getInstance(); diff --git a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java index e9f0dda0fba16..8f7706426c362 100644 --- a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java @@ -402,6 +402,9 @@ public final class PipeMessages { public static final String SUBTASK_CLOSE_FAILED = "Failed to close the subtask {}."; public static final String LOG_FAILED_TO_START_THE_SUBTASK_DD3FC1BA = "Failed to start the subtask {}."; + public static final String + EXCEPTION_THE_SUBTASK_CAN_NOT_BE_STARTED_BECAUSE_THE_EXECUTOR_HAS_BEEN_SHUT_DOWN_9989B212 = + "The subtask can not be started because the executor has been shut down."; // ===================== PipeReportableSubtask ===================== diff --git a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java index 8e09d70739e14..04a88163523fd 100644 --- a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java @@ -385,6 +385,9 @@ public final class PipeMessages { public static final String SUBTASK_CLOSE_FAILED = "关闭子任务 {} 失败。"; public static final String LOG_FAILED_TO_START_THE_SUBTASK_DD3FC1BA = "启动子任务 {} 失败。"; + public static final String + EXCEPTION_THE_SUBTASK_CAN_NOT_BE_STARTED_BECAUSE_THE_EXECUTOR_HAS_BEEN_SHUT_DOWN_9989B212 = + "子任务无法启动，因为执行器已关闭。"; // ===================== PipeReportableSubtask ===================== diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/execution/PipeSubtaskExecutor.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/execution/PipeSubtaskExecutor.java index afa3b33b0000f..dc41b59651b70 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/execution/PipeSubtaskExecutor.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/execution/PipeSubtaskExecutor.java @@ -38,6 +38,7 @@ import java.util.Objects; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; @@ -139,6 +140,17 @@ public final synchronized void start(final String subTaskID) { } final PipeSubtask subtask = registeredIdSubtaskMapper.get(subTaskID); + if (isShutdown()) { + // A shut-down executor can never execute the subtask. Fail fast before touching the + // "submitting self" state, otherwise start() may report a false success and the subtask + // will be counted as running but never executed until the node restarts. + LOGGER.error( + PipeMessages.LOG_FAILED_TO_START_THE_SUBTASK_DD3FC1BA, subtask.getDisplayTaskID()); + throw new RejectedExecutionException( + PipeMessages + .EXCEPTION_THE_SUBTASK_CAN_NOT_BE_STARTED_BECAUSE_THE_EXECUTOR_HAS_BEEN_SHUT_DOWN_9989B212); + } + if (subtask.isSubmittingSelf()) { if (LOGGER.isDebugEnabled()) { LOGGER.debug(PipeMessages.SUBTASK_ALREADY_RUNNING, subtask.getDisplayTaskID()); diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java index 95863a79da9c3..ff21e92835da7 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java @@ -84,15 +84,6 @@ public void bindExecutors( this.subtaskScheduler = subtaskScheduler; } - @Override - protected synchronized void onDisallowSubmittingSelf() { - // The subtask is not submitting itself any more. The "isSubmitted" sign is cleared so that a - // later allowSubmittingSelf() + submitSelf() can really submit the subtask again, even if the - // previous submission was never dispatched to the worker pool, e.g., because the submission was - // rejected or scheduled on an executor that has been shut down. - isSubmitted = false; - } - @Override public void onSuccess(final Boolean hasAtLeastOneEventProcessed) { preScheduleLowPriorityTask(100); From 8eeb62e271166ce0c34785dfd38ac2f319c6a6b4 Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Wed, 4 Nov 2026 15:00:20 +0800 Subject: [PATCH 5/5] Shorten comments and document the intent of the new tests Replace the verbose inline comments in the pipe task start fix with concise ones, and state the scenario and expected behavior in a short Javadoc above each new test. --- .../execution/PipeSubtaskExecutorManager.java | 5 +-- .../agent/task/PipeDataNodeTaskAgentTest.java | 8 ++-- .../task/PipeSinkSubtaskExecutorTest.java | 37 +++++++++---------- .../agent/task/PipeSubtaskExecutorTest.java | 6 ++- .../PipeSubtaskExecutorManagerTest.java | 11 +++--- .../pipe/agent/task/PipeTaskAgent.java | 8 ++-- .../task/execution/PipeSubtaskExecutor.java | 5 +-- .../task/subtask/PipeAbstractSinkSubtask.java | 6 +-- 8 files changed, 40 insertions(+), 46 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/execution/PipeSubtaskExecutorManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/execution/PipeSubtaskExecutorManager.java index 36a8602dea15c..654726912b193 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/execution/PipeSubtaskExecutorManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/execution/PipeSubtaskExecutorManager.java @@ -59,9 +59,8 @@ public synchronized IoTConsensusV2SubtaskExecutor getConsensusExecutor() { return (IoTConsensusV2SubtaskExecutor) consensusExecutor; } - // The consensus executor is shared by all the consensus pipes of the node. If it has been shut - // down unexpectedly, recreate it, otherwise all the consensus pipes created or restarted - // afterwards would never be able to submit their sink subtasks. + // The consensus executor is shared by all consensus pipes. Recreate it if it has been shut + // down, otherwise the following consensus pipes can never start their sink subtasks. LOGGER.warn( DataNodePipeMessages .LOG_THE_SHARED_CONSENSUS_EXECUTOR_HAS_BEEN_SHUT_DOWN_RECREATING_IT_C14080E3); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgentTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgentTest.java index 4e5ccca3a5009..dde64b4ef9360 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgentTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgentTest.java @@ -579,6 +579,10 @@ public void testCarryOverRequiresStableLeaderAndLocalTask() { .getProgressIndex()); } + /** + * A task that fails to start is dropped and unregistered, so the following meta sync creates it + * again. + */ @Test public void testFailedTaskStartIsRolledBackForRetry() throws Exception { final PipeDataNodeTaskAgent agent = newActiveTaskAgent(); @@ -597,8 +601,6 @@ public void testFailedTaskStartIsRolledBackForRetry() throws Exception { pipeMeta.getRuntimeMeta().getStatus().set(PipeStatus.RUNNING); keeper.addPipeMeta(pipeMeta); - // A task which has been created but can not be started, e.g., the sink of the task has been - // rejected by a shut-down thread pool. final PipeTask failedTask = Mockito.mock(PipeTask.class); Mockito.doThrow(new PipeException("failed to start the sink")).when(failedTask).start(); manager.addPipeTask(pipeMeta.getStaticMeta(), REGION_ID, failedTask); @@ -608,8 +610,6 @@ public void testFailedTaskStartIsRolledBackForRetry() throws Exception { startPipeTask.setAccessible(true); startPipeTask.invoke(agent, REGION_ID, pipeMeta.getStaticMeta()); - // The half-started task is dropped and its task meta is removed, so that the following meta - // sync creates and starts the task again instead of keeping it as "RUNNING but never runs". Mockito.verify(failedTask).drop(); Assert.assertNull(manager.getPipeTask(pipeMeta.getStaticMeta(), REGION_ID)); Assert.assertNull(pipeMeta.getRuntimeMeta().getConsensusGroupId2TaskMetaMap().get(REGION_ID)); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeSinkSubtaskExecutorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeSinkSubtaskExecutorTest.java index 7a97f029fb7ad..bb24ac2a7792c 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeSinkSubtaskExecutorTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeSinkSubtaskExecutorTest.java @@ -58,14 +58,16 @@ public void setUp() throws Exception { mock(PipeConnector.class))); } + /** + * A shut-down executor must reject {@code start()} and must not leave the subtask in the + * "submitting" state. + */ @Test public void testStartAfterShutdown() { executor.register(subtask); executor.shutdown(); Assert.assertTrue(executor.isShutdown()); - // The submission is rejected by the shut-down executor, and the rejection must be reported - // instead of being swallowed as "started successfully"... for (int i = 0; i < 2; ++i) { try { executor.start(subtask.getTaskID()); @@ -73,19 +75,17 @@ public void testStartAfterShutdown() { } catch (final RejectedExecutionException e) { // expected } - // ... and the failed submission must not leave the subtask in the "submitting" state. Assert.assertFalse(subtask.isSubmittingSelf()); } } + /** A rejected self-submission must be reported instead of being swallowed. */ @Test public void testSubmitSelfAfterShutdown() throws Exception { executor.register(subtask); executor.shutdown(); allowSubmittingSelfAgain(subtask); - // The subtask submits itself again after each round, so a rejected submission must be reported - // instead of being swallowed. Otherwise the subtask would stop transferring events silently. for (int i = 0; i < 2; ++i) { try { subtask.submitSelf(); @@ -96,6 +96,10 @@ public void testSubmitSelfAfterShutdown() throws Exception { } } + /** + * A rejected delayed submission must clear the submitted flag, otherwise the following + * submissions would be skipped. + */ @Test public void testDelayedSubmitSelfAfterShutdown() throws Exception { final SinkWithSchedulingDelay sink = mock(SinkWithSchedulingDelay.class); @@ -116,9 +120,6 @@ public void testDelayedSubmitSelfAfterShutdown() throws Exception { executor.shutdown(); allowSubmittingSelfAgain(delayedSubtask); - // A delayed submission marks the subtask as submitted before scheduling it. If the scheduling - // is rejected, the sign must be cleared as well, otherwise every following submission would be - // skipped and the subtask would never be submitted again. for (int i = 0; i < 2; ++i) { try { delayedSubtask.submitSelf(); @@ -129,6 +130,10 @@ public void testDelayedSubmitSelfAfterShutdown() throws Exception { } } + /** + * Stopping and restarting a subtask while its previous call is still running must not start a + * second concurrent call. + */ @Test public void testStartWhileThePreviousCallIsStillRunning() throws Exception { final AtomicInteger concurrentCallCount = new AtomicInteger(0); @@ -169,8 +174,6 @@ protected boolean executeOnce() { executor.start(blockedSubtask.getTaskID()); Assert.assertTrue(firstCallEntered.await(5, TimeUnit.SECONDS)); - // Stop and start while the first call is still running. The subtask must not be submitted - // again, because a PipeSubtask is not designed to run in parallel with itself. executor.stop(blockedSubtask.getTaskID()); executor.start(blockedSubtask.getTaskID()); @@ -185,6 +188,10 @@ protected boolean executeOnce() { } } + /** + * {@code start()} must fail after shutdown even if a pending delayed submission already marked + * the subtask as submitted. + */ @Test public void testStartAfterShutdownWithPendingDelayedSubmission() { final SinkWithSchedulingDelay sink = mock(SinkWithSchedulingDelay.class); @@ -201,13 +208,8 @@ public void testStartAfterShutdownWithPendingDelayedSubmission() { sink); executor.register(delayedSubtask); - - // The delayed submission is accepted and marks the subtask as submitted before it is - // dispatched. executor.start(delayedSubtask.getTaskID()); - // The delayed submission will never be dispatched to the shut-down worker pool, so a following - // start() must not take the "already submitted" branch and report a false success. executor.shutdown(); try { executor.start(delayedSubtask.getTaskID()); @@ -217,10 +219,7 @@ public void testStartAfterShutdownWithPendingDelayedSubmission() { } } - /** - * Simulates the race where a subtask is submitting itself while its executor is being shut down: - * the executor can not submit it any more, but the subtask has not been told to stop yet. - */ + /** Clears the stop flag to simulate a subtask that has not been told to stop yet. */ private void allowSubmittingSelfAgain(final PipeSubtask subtask) throws Exception { final Field shouldStopSubmittingSelfField = PipeSubtask.class.getDeclaredField("shouldStopSubmittingSelf"); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeSubtaskExecutorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeSubtaskExecutorTest.java index 88838cbd75d98..ad12f4871a7cf 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeSubtaskExecutorTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeSubtaskExecutorTest.java @@ -118,6 +118,10 @@ public void testStop() { Assert.assertFalse(subtask.isSubmittingSelf()); } + /** + * A rejected submission must be rolled back, otherwise the following start() calls would take the + * "already running" branch and report a false success. + */ @Test public void testStartWithRejectedSubmission() { executor.register(subtask); @@ -130,8 +134,6 @@ public void testStartWithRejectedSubmission() { // expected } - // The failed submission must be rolled back, otherwise the following start calls would take - // the "already running" branch and report a false success. verify(subtask).allowSubmittingSelf(); verify(subtask).disallowSubmittingSelf(); Assert.assertFalse(executor.isShutdown()); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/execution/PipeSubtaskExecutorManagerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/execution/PipeSubtaskExecutorManagerTest.java index 21267f305666b..9d1375fdf9588 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/execution/PipeSubtaskExecutorManagerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/execution/PipeSubtaskExecutorManagerTest.java @@ -29,25 +29,26 @@ public class PipeSubtaskExecutorManagerTest { @After public void tearDown() { - // Do not leave the node-wide shared consensus executor shut down for the other tests in the - // same JVM. getConsensusExecutor() recreates it if it has been shut down. + // Recreate the shared consensus executor so the following tests are not affected. PipeSubtaskExecutorManager.getInstance().getConsensusExecutor(); } + /** The consensus executor is a singleton while it is alive. */ @Test public void testGetConsensusExecutor() { final PipeSubtaskExecutorManager manager = PipeSubtaskExecutorManager.getInstance(); final IoTConsensusV2SubtaskExecutor consensusExecutor = manager.getConsensusExecutor(); Assert.assertFalse(consensusExecutor.isShutdown()); - // The consensus executor is shared by all the consensus pipes, so it must be a singleton as - // long as it is alive. Assert.assertSame(consensusExecutor, manager.getConsensusExecutor()); consensusExecutor.shutdown(); Assert.assertTrue(consensusExecutor.isShutdown()); } + /** + * A shut-down consensus executor is recreated so that consensus pipes can start their subtasks. + */ @Test public void testGetConsensusExecutorRecoversFromShutdown() { final PipeSubtaskExecutorManager manager = PipeSubtaskExecutorManager.getInstance(); @@ -56,8 +57,6 @@ public void testGetConsensusExecutorRecoversFromShutdown() { consensusExecutor.shutdown(); Assert.assertTrue(consensusExecutor.isShutdown()); - // A shut-down consensus executor must not be handed to the following consensus pipes, - // otherwise none of their sink subtasks could ever be started. final IoTConsensusV2SubtaskExecutor recoveredExecutor = manager.getConsensusExecutor(); Assert.assertNotSame(consensusExecutor, recoveredExecutor); Assert.assertFalse(recoveredExecutor.isShutdown()); diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java index e05cb78fb3cf5..6130464a83ebc 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java @@ -991,11 +991,9 @@ private void startPipeTask(final int consensusGroupId, final PipeStaticMeta pipe pipeStaticMeta.getPipeName(), e); - // The task is not running now. Dropping it instead of keeping it as is, because a task that - // has already been created is never started again by the following meta sync, which would - // leave the pipe alive in the metadata but dead in fact until the node restarts. After the - // task meta is removed here, the following meta sync sees no task meta of this consensus - // group and creates the task again, which gives the task another chance to start. + // A created task is never started again by the following meta sync. Drop it and remove its + // task meta so that the meta sync creates and starts the task again instead of keeping it + // "RUNNING" but dead until the node restarts. rollbackPipeTask(consensusGroupId, pipeStaticMeta); return; } diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/execution/PipeSubtaskExecutor.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/execution/PipeSubtaskExecutor.java index dc41b59651b70..d2ac504002c92 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/execution/PipeSubtaskExecutor.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/execution/PipeSubtaskExecutor.java @@ -141,9 +141,8 @@ public final synchronized void start(final String subTaskID) { final PipeSubtask subtask = registeredIdSubtaskMapper.get(subTaskID); if (isShutdown()) { - // A shut-down executor can never execute the subtask. Fail fast before touching the - // "submitting self" state, otherwise start() may report a false success and the subtask - // will be counted as running but never executed until the node restarts. + // Fail fast before touching the "submitting self" state, otherwise start() may report a + // false success and the subtask will never be executed. LOGGER.error( PipeMessages.LOG_FAILED_TO_START_THE_SUBTASK_DD3FC1BA, subtask.getDisplayTaskID()); throw new RejectedExecutionException( diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java index ff21e92835da7..c7d5b84cc96f1 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java @@ -293,10 +293,8 @@ public synchronized void submitSelf() { submitSelfToWorker(true); } catch (final RejectedExecutionException e) { - // The worker pool or the scheduler has been shut down, e.g., because the pipe has been - // dropped. The "isSubmitted" sign must be cleared, otherwise the subtask would never be - // submitted again and would silently stop transferring events. The rejection is reported - // instead of being swallowed by the callback executor. + // The executor has been shut down, e.g., because the pipe was dropped. Clear isSubmitted so + // the subtask can be submitted again, and report the rejection. isSubmitted = false; LOGGER.warn( PipeMessages