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..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 @@ -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,22 @@ 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 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); + 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/PipeDataNodeTaskAgentTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgentTest.java index fe1bfbe4aabe8..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 @@ -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()); } + /** + * 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(); + 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); + + 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()); + + 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/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..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 @@ -20,13 +20,24 @@ 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; +import org.junit.Assert; import org.junit.Before; +import org.junit.Test; 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; public class PipeSinkSubtaskExecutorTest extends PipeSubtaskExecutorTest { @@ -46,4 +57,175 @@ public void setUp() throws Exception { mock(UnboundedBlockingPendingQueue.class), 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()); + + 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 + } + 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); + + 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 + } + } + } + + /** + * 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); + 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); + + 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 + } + } + } + + /** + * 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); + 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)); + + 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(); + } + } + + /** + * {@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); + 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); + executor.start(delayedSubtask.getTaskID()); + + 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 + } + } + + /** 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"); + shouldStopSubmittingSelfField.setAccessible(true); + ((AtomicBoolean) shouldStopSubmittingSelfField.get(subtask)).set(false); + } + + private interface SinkWithSchedulingDelay extends PipeConnector, PipeSinkWithSchedulingDelay {} } 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..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 @@ -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,27 @@ 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); + 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 + } + + 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..9d1375fdf9588 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/execution/PipeSubtaskExecutorManagerTest.java @@ -0,0 +1,66 @@ +/* + * 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.After; +import org.junit.Assert; +import org.junit.Test; + +public class PipeSubtaskExecutorManagerTest { + + @After + public void tearDown() { + // 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()); + 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(); + + final IoTConsensusV2SubtaskExecutor consensusExecutor = manager.getConsensusExecutor(); + consensusExecutor.shutdown(); + Assert.assertTrue(consensusExecutor.isShutdown()); + + 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..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 @@ -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 ===================== @@ -389,7 +396,15 @@ 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 {}."; + 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 47ddd88162ddc..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 @@ -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 ===================== @@ -373,7 +379,15 @@ 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 = + "启动子任务 {} 失败。"; + 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/PipeTaskAgent.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java index 8352d2c6be1fb..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 @@ -977,9 +977,40 @@ 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); + + // 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; + } + + 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); } } 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..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 @@ -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,13 +140,37 @@ public final synchronized void start(final String subTaskID) { } final PipeSubtask subtask = registeredIdSubtaskMapper.get(subTaskID); + if (isShutdown()) { + // 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( + 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()); } } 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..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 @@ -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 { @@ -279,17 +280,29 @@ 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 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 + .LOG_FAILED_TO_SUBMIT_THE_SUBTASK_ARG_BECAUSE_THE_EXECUTOR_HAS_BEEN_SHUT_DOWN_198335BE, + getDisplayTaskID(), + e); + throw e; + } } @Override