From f0ee3773e1f9bd7726a916e41e7357e6486ecd20 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 7 Aug 2026 14:45:16 +0800 Subject: [PATCH] fix(pipe): reuse safe local progress on alter --- .../agent/task/PipeDataNodeTaskAgent.java | 86 ++++++++++ .../impl/DataNodeInternalRPCServiceImpl.java | 17 ++ .../thrift/impl/PushMultiPipeMetaHelper.java | 29 +++- .../agent/task/PipeDataNodeTaskAgentTest.java | 159 ++++++++++++++++++ ...alRPCServiceImplPushMultiPipeMetaTest.java | 72 ++++++++ 5 files changed, 358 insertions(+), 5 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java index 3adad062db7dd..07220c57cbabc 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java @@ -38,6 +38,7 @@ import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta; import org.apache.iotdb.commons.pipe.agent.task.meta.PipeRuntimeMeta; import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta; +import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStatus; import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta; import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTemporaryMeta; import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTemporaryMetaInAgent; @@ -102,6 +103,7 @@ import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.BiPredicate; import java.util.function.Consumer; import java.util.stream.Collectors; @@ -199,6 +201,8 @@ public List handlePipeMetaChangesInternal( return Collections.emptyList(); } + carryOverLocalProgressIndexForAlter(pipeMetaListFromCoordinator); + final List exceptionMessages = super.handlePipeMetaChangesInternal(pipeMetaListFromCoordinator); @@ -217,6 +221,88 @@ public List handlePipeMetaChangesInternal( return exceptionMessages; } + /** + * Carry the committed progress of an old local task into an altered task when it is safe to do + * so. The old task is dropped before the new task is created, therefore this must run before + * {@link PipeTaskAgent#handlePipeMetaChangesInternal(List)} starts applying the metadata list. + * + *

We deliberately only carry progress when the old and new task stay on this DataNode and + * their realtime-only modes are unchanged. Mode changes have explicit progress semantics in the + * ConfigNode metadata (for example, realtime-only to historical resets to {@code + * MinimumProgressIndex}), and leader changes must use the coordinator checkpoint because the old + * task is not local to the new leader. + */ + private void carryOverLocalProgressIndexForAlter( + final List pipeMetaListFromCoordinator) { + for (final PipeMeta droppedPipeMeta : pipeMetaListFromCoordinator) { + if (droppedPipeMeta.getRuntimeMeta().getStatus().get() != PipeStatus.DROPPED) { + continue; + } + + final PipeStaticMeta oldStaticMeta = droppedPipeMeta.getStaticMeta(); + final PipeMeta localOldPipeMeta = pipeMetaKeeper.getPipeMeta(oldStaticMeta); + if (localOldPipeMeta == null) { + continue; + } + + for (final PipeMeta updatedPipeMeta : pipeMetaListFromCoordinator) { + if (updatedPipeMeta == droppedPipeMeta + || updatedPipeMeta.getRuntimeMeta().getStatus().get() == PipeStatus.DROPPED + || !oldStaticMeta.getPipeName().equals(updatedPipeMeta.getStaticMeta().getPipeName()) + || oldStaticMeta.visibleUnderTableModel() + != updatedPipeMeta.getStaticMeta().visibleUnderTableModel()) { + continue; + } + + carryOverLocalProgressIndexForAlter( + oldStaticMeta, + localOldPipeMeta, + updatedPipeMeta, + CONFIG.getDataNodeId(), + (staticMeta, consensusGroupId) -> + pipeTaskManager.getPipeTask(staticMeta, consensusGroupId) != null); + } + } + } + + static void carryOverLocalProgressIndexForAlter( + final PipeStaticMeta oldStaticMeta, + final PipeMeta localOldPipeMeta, + final PipeMeta updatedPipeMeta, + final int localNodeId, + final BiPredicate localTaskExists) { + final PipeStaticMeta updatedStaticMeta = updatedPipeMeta.getStaticMeta(); + + // A mode change has an explicit cutover/reset meaning in ConfigNode. In particular, a + // realtime-only -> historical alter must retain MinimumProgressIndex to scan old files. + if (PipeTaskAgent.isRealtimeOnlyPipe(oldStaticMeta.getSourceParameters()) + != PipeTaskAgent.isRealtimeOnlyPipe(updatedStaticMeta.getSourceParameters())) { + return; + } + + final Map localTaskMetaMap = + localOldPipeMeta.getRuntimeMeta().getConsensusGroupId2TaskMetaMap(); + final Map updatedTaskMetaMap = + updatedPipeMeta.getRuntimeMeta().getConsensusGroupId2TaskMetaMap(); + + for (final Map.Entry entry : updatedTaskMetaMap.entrySet()) { + final int consensusGroupId = entry.getKey(); + final PipeTaskMeta updatedTaskMeta = entry.getValue(); + final PipeTaskMeta localTaskMeta = localTaskMetaMap.get(consensusGroupId); + + // Only the old task's actual leader owns an authoritative local checkpoint. Requiring the + // new task to stay on the same node also avoids losing the checkpoint during leader change. + if (localTaskMeta == null + || localTaskMeta.getLeaderNodeId() != localNodeId + || updatedTaskMeta.getLeaderNodeId() != localNodeId + || !localTaskExists.test(oldStaticMeta, consensusGroupId)) { + continue; + } + + updatedTaskMeta.updateProgressIndex(localTaskMeta.getProgressIndex()); + } + } + private Set clearSchemaRegionListeningQueueIfNecessary( final List pipeMetaListFromCoordinator) throws IllegalPathException { final Map schemaRegionId2ListeningQueueNewFirstIndex = new HashMap<>(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java index a06d10413fd42..4bb2a50d1f23d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java @@ -1403,6 +1403,23 @@ public TPushPipeMetaRespExceptionMessage handleDropPipe(final String pipeName) { return PipeDataNodeAgent.task().handleDropPipe(pipeName); } + @Override + public boolean handlePipeMetaChanges( + final List pipeMetas, + final List exceptionMessages) { + final List exceptionMessagesFromAgent = + PipeDataNodeAgent.task() + .handlePipeMetaChanges( + pipeMetas.stream() + .map(PipeMeta::deserialize4TaskAgent) + .collect(Collectors.toList())); + if (exceptionMessagesFromAgent == null) { + return false; + } + exceptionMessages.addAll(exceptionMessagesFromAgent); + return true; + } + @Override public TPushPipeMetaRespExceptionMessage handleSinglePipeMeta(final ByteBuffer pipeMeta) { return PipeDataNodeAgent.task() diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/PushMultiPipeMetaHelper.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/PushMultiPipeMetaHelper.java index b96960f3e0100..e91108edb30ce 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/PushMultiPipeMetaHelper.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/PushMultiPipeMetaHelper.java @@ -46,6 +46,27 @@ interface Handler { TPushPipeMetaRespExceptionMessage handleDropPipe(String pipeName) throws Exception; TPushPipeMetaRespExceptionMessage handleSinglePipeMeta(ByteBuffer pipeMeta) throws Exception; + + /** + * Handles all pipe metadata in one agent invocation. Alter pipe sends the old dropped metadata + * and the new metadata together, so they must be visible to the agent at the same time when it + * decides whether the old task's local progress can be reused. + * + *

The default implementation preserves the per-metadata behavior for handlers that do not + * need batch processing. + */ + default boolean handlePipeMetaChanges( + final List pipeMetas, + final List exceptionMessages) + throws Exception { + for (final ByteBuffer pipeMeta : pipeMetas) { + final TPushPipeMetaRespExceptionMessage message = handleSinglePipeMeta(pipeMeta); + if (message != null) { + exceptionMessages.add(message); + } + } + return true; + } } static TPushPipeMetaResp pushMultiPipeMeta( @@ -60,11 +81,9 @@ static TPushPipeMetaResp pushMultiPipeMeta( } } } else if (req.isSetPipeMetas()) { - for (final ByteBuffer pipeMeta : req.getPipeMetas()) { - final TPushPipeMetaRespExceptionMessage message = handler.handleSinglePipeMeta(pipeMeta); - if (message != null) { - exceptionMessages.add(message); - } + if (!handler.handlePipeMetaChanges(req.getPipeMetas(), exceptionMessages)) { + return new TPushPipeMetaResp() + .setStatus(new TSStatus(TSStatusCode.PIPE_PUSH_META_TIMEOUT.getStatusCode())); } } else { throw new Exception(DataNodeMiscMessages.INVALID_PUSH_MULTI_PIPE_META_REQ); 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 d3249933aa114..3dd93e877890c 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 @@ -20,9 +20,14 @@ package org.apache.iotdb.db.pipe.agent.task; import org.apache.iotdb.commons.conf.CommonDescriptor; +import org.apache.iotdb.commons.consensus.index.ProgressIndex; +import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex; +import org.apache.iotdb.commons.consensus.index.impl.SimpleProgressIndex; import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta; import org.apache.iotdb.commons.pipe.agent.task.meta.PipeRuntimeMeta; import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta; +import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta; +import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant; import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent; import org.apache.iotdb.pipe.api.exception.PipeException; @@ -30,9 +35,15 @@ import org.junit.Test; import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; public class PipeDataNodeTaskAgentTest { + private static final int LOCAL_NODE_ID = 1; + private static final int REGION_ID = 7; + @Test public void testCreateMemoryCheckStillRunsWhenNoPipeTasksNeedToBeCreated() throws Exception { final boolean originalPipeEnableMemoryCheck = @@ -68,4 +79,152 @@ public void testCreateMemoryCheckStillRunsWhenNoPipeTasksNeedToBeCreated() throw .setPipeTotalFloatingMemoryProportion(originalPipeTotalFloatingMemoryProportion); } } + + @Test + public void testCarryOverCommittedProgressForResumeAlter() { + final PipeStaticMeta oldStaticMeta = createStaticMeta(1, false); + final PipeStaticMeta updatedStaticMeta = createStaticMeta(2, false); + final PipeMeta localOldPipeMeta = + createPipeMeta(oldStaticMeta, new SimpleProgressIndex(1, 20L), LOCAL_NODE_ID); + final PipeMeta updatedPipeMeta = + createPipeMeta(updatedStaticMeta, new SimpleProgressIndex(1, 10L), LOCAL_NODE_ID); + + PipeDataNodeTaskAgent.carryOverLocalProgressIndexForAlter( + oldStaticMeta, + localOldPipeMeta, + updatedPipeMeta, + LOCAL_NODE_ID, + (staticMeta, regionId) -> regionId == REGION_ID); + + Assert.assertEquals( + new SimpleProgressIndex(1, 20L), + updatedPipeMeta + .getRuntimeMeta() + .getConsensusGroupId2TaskMetaMap() + .get(REGION_ID) + .getProgressIndex()); + } + + @Test + public void testCarryOverDoesNotOverrideCoordinatorProgress() { + final PipeStaticMeta oldStaticMeta = createStaticMeta(1, false); + final PipeStaticMeta updatedStaticMeta = createStaticMeta(2, false); + final PipeMeta localOldPipeMeta = + createPipeMeta(oldStaticMeta, new SimpleProgressIndex(1, 10L), LOCAL_NODE_ID); + final PipeMeta updatedPipeMeta = + createPipeMeta(updatedStaticMeta, new SimpleProgressIndex(1, 20L), LOCAL_NODE_ID); + + PipeDataNodeTaskAgent.carryOverLocalProgressIndexForAlter( + oldStaticMeta, + localOldPipeMeta, + updatedPipeMeta, + LOCAL_NODE_ID, + (staticMeta, regionId) -> true); + + Assert.assertEquals( + new SimpleProgressIndex(1, 20L), + updatedPipeMeta + .getRuntimeMeta() + .getConsensusGroupId2TaskMetaMap() + .get(REGION_ID) + .getProgressIndex()); + } + + @Test + public void testCarryOverDoesNotOverrideProgressResetOnModeChange() { + final PipeStaticMeta oldStaticMeta = createStaticMeta(1, false); + final PipeStaticMeta updatedStaticMeta = createStaticMeta(2, true); + final PipeMeta localOldPipeMeta = + createPipeMeta(oldStaticMeta, new SimpleProgressIndex(1, 20L), LOCAL_NODE_ID); + final PipeMeta updatedPipeMeta = + createPipeMeta(updatedStaticMeta, MinimumProgressIndex.INSTANCE, LOCAL_NODE_ID); + + PipeDataNodeTaskAgent.carryOverLocalProgressIndexForAlter( + oldStaticMeta, + localOldPipeMeta, + updatedPipeMeta, + LOCAL_NODE_ID, + (staticMeta, regionId) -> true); + + Assert.assertSame( + MinimumProgressIndex.INSTANCE, + updatedPipeMeta + .getRuntimeMeta() + .getConsensusGroupId2TaskMetaMap() + .get(REGION_ID) + .getProgressIndex()); + } + + @Test + public void testCarryOverRequiresStableLeaderAndLocalTask() { + final PipeStaticMeta oldStaticMeta = createStaticMeta(1, false); + final PipeStaticMeta updatedStaticMeta = createStaticMeta(2, false); + final PipeMeta localOldPipeMeta = + createPipeMeta(oldStaticMeta, new SimpleProgressIndex(1, 20L), LOCAL_NODE_ID); + + final PipeMeta localOldPipeMetaWithLeaderChange = + createPipeMeta(oldStaticMeta, new SimpleProgressIndex(1, 20L), 2); + final PipeMeta updatedWithOldLeaderChange = + createPipeMeta(updatedStaticMeta, new SimpleProgressIndex(1, 10L), LOCAL_NODE_ID); + PipeDataNodeTaskAgent.carryOverLocalProgressIndexForAlter( + oldStaticMeta, + localOldPipeMetaWithLeaderChange, + updatedWithOldLeaderChange, + LOCAL_NODE_ID, + (staticMeta, regionId) -> true); + Assert.assertEquals( + new SimpleProgressIndex(1, 10L), + updatedWithOldLeaderChange + .getRuntimeMeta() + .getConsensusGroupId2TaskMetaMap() + .get(REGION_ID) + .getProgressIndex()); + + final PipeMeta updatedWithLeaderChange = + createPipeMeta(updatedStaticMeta, new SimpleProgressIndex(1, 10L), 2); + PipeDataNodeTaskAgent.carryOverLocalProgressIndexForAlter( + oldStaticMeta, + localOldPipeMeta, + updatedWithLeaderChange, + LOCAL_NODE_ID, + (staticMeta, regionId) -> true); + Assert.assertEquals( + new SimpleProgressIndex(1, 10L), + updatedWithLeaderChange + .getRuntimeMeta() + .getConsensusGroupId2TaskMetaMap() + .get(REGION_ID) + .getProgressIndex()); + + final PipeMeta updatedWithoutLocalTask = + createPipeMeta(updatedStaticMeta, new SimpleProgressIndex(1, 10L), LOCAL_NODE_ID); + PipeDataNodeTaskAgent.carryOverLocalProgressIndexForAlter( + oldStaticMeta, + localOldPipeMeta, + updatedWithoutLocalTask, + LOCAL_NODE_ID, + (staticMeta, regionId) -> false); + Assert.assertEquals( + new SimpleProgressIndex(1, 10L), + updatedWithoutLocalTask + .getRuntimeMeta() + .getConsensusGroupId2TaskMetaMap() + .get(REGION_ID) + .getProgressIndex()); + } + + private PipeStaticMeta createStaticMeta(final long creationTime, final boolean historyEnabled) { + final Map sourceAttributes = new HashMap<>(); + sourceAttributes.put( + PipeSourceConstant.SOURCE_HISTORY_ENABLE_KEY, Boolean.toString(historyEnabled)); + return new PipeStaticMeta( + String.valueOf('p'), creationTime, sourceAttributes, new HashMap<>(), new HashMap<>()); + } + + private PipeMeta createPipeMeta( + final PipeStaticMeta staticMeta, final ProgressIndex progressIndex, final int leaderId) { + final ConcurrentMap taskMetaMap = new ConcurrentHashMap<>(); + taskMetaMap.put(REGION_ID, new PipeTaskMeta(progressIndex, leaderId)); + return new PipeMeta(staticMeta, new PipeRuntimeMeta(taskMetaMap)); + } } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImplPushMultiPipeMetaTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImplPushMultiPipeMetaTest.java index 0ae284bce9d7a..8dc6523d8ed75 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImplPushMultiPipeMetaTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImplPushMultiPipeMetaTest.java @@ -30,6 +30,7 @@ import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.List; import java.util.concurrent.atomic.AtomicInteger; @@ -240,6 +241,77 @@ public TPushPipeMetaRespExceptionMessage handleSinglePipeMeta( Assert.assertEquals(0, resp.getExceptionMessagesSize()); } + @Test + public void testPushMultiPipeMetaInvokesBatchHandlerOnce() { + final AtomicInteger batchCallCount = new AtomicInteger(0); + final AtomicInteger singleCallCount = new AtomicInteger(0); + final TPushPipeMetaResp resp = + PushMultiPipeMetaHelper.pushMultiPipeMeta( + new TPushMultiPipeMetaReq() + .setPipeMetas( + Arrays.asList( + ByteBuffer.wrap(new byte[] {1}), ByteBuffer.wrap(new byte[] {2}))), + new PushMultiPipeMetaHelper.Handler() { + @Override + public TPushPipeMetaRespExceptionMessage handleDropPipe(final String pipeName) { + Assert.fail("Unexpected drop pipe request"); + return null; + } + + @Override + public boolean handlePipeMetaChanges( + final List pipeMetas, + final List exceptionMessages) { + batchCallCount.incrementAndGet(); + Assert.assertEquals(2, pipeMetas.size()); + return true; + } + + @Override + public TPushPipeMetaRespExceptionMessage handleSinglePipeMeta( + final ByteBuffer pipeMeta) { + singleCallCount.incrementAndGet(); + return null; + } + }); + + Assert.assertEquals(1, batchCallCount.get()); + Assert.assertEquals(0, singleCallCount.get()); + Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), resp.getStatus().getCode()); + } + + @Test + public void testPushMultiPipeMetaReturnsTimeoutWhenBatchHandlerTimesOut() { + final TPushPipeMetaResp resp = + PushMultiPipeMetaHelper.pushMultiPipeMeta( + new TPushMultiPipeMetaReq() + .setPipeMetas(Collections.singletonList(ByteBuffer.wrap(new byte[] {1}))), + new PushMultiPipeMetaHelper.Handler() { + @Override + public TPushPipeMetaRespExceptionMessage handleDropPipe(final String pipeName) { + Assert.fail("Unexpected drop pipe request"); + return null; + } + + @Override + public boolean handlePipeMetaChanges( + final List pipeMetas, + final List exceptionMessages) { + return false; + } + + @Override + public TPushPipeMetaRespExceptionMessage handleSinglePipeMeta( + final ByteBuffer pipeMeta) { + Assert.fail("Unexpected single pipe meta request"); + return null; + } + }); + + Assert.assertEquals( + TSStatusCode.PIPE_PUSH_META_TIMEOUT.getStatusCode(), resp.getStatus().getCode()); + } + private static TPushPipeMetaRespExceptionMessage newExceptionMessage(final String pipeName) { return new TPushPipeMetaRespExceptionMessage(pipeName, "failed", 1L); }