This is an automated email from the ASF dual-hosted git repository. Caideyipi pushed a commit to branch fix/pipe-partial-metadata-preserve-tasks in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 0aee24ecd211505d659b7fe7f7a4b93e05e48c65 Author: Caideyipi <[email protected]> AuthorDate: Wed Oct 28 19:35:35 2026 +0800 Fix partial pipe metadata reconciliation --- .../db/pipe/agent/task/PipeDataNodeTaskAgent.java | 32 ++++++ .../impl/DataNodeInternalRPCServiceImpl.java | 2 +- .../pipe/agent/task/PipeDataNodeTaskAgentTest.java | 109 +++++++++++++++++++++ 3 files changed, 142 insertions(+), 1 deletion(-) 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 359bdd79c50..a95b8ab410e 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 @@ -199,6 +199,38 @@ public class PipeDataNodeTaskAgent extends PipeTaskAgent { .put(consensusGroupId, pipeTaskMeta); } + /** Applies a partial metadata batch without dropping pipes omitted from the request. */ + public List<TPushPipeMetaRespExceptionMessage> handleMultiPipeMetaChanges( + final List<PipeMeta> pipeMetaListFromCoordinator) { + if (pipeMetaListFromCoordinator.isEmpty()) { + return Collections.emptyList(); + } + if (!tryWriteLockWithTimeOutInMs( + CommonDescriptor.getInstance().getConfig().getDnConnectionTimeoutInMS() * 2L / 3)) { + return null; + } + try { + // Reconciliation and alter-progress reuse need one complete view under the same write lock. + // Resolve replacements using the keeper's name/visibility identity, including new versions + // of existing pipes. Preserve the incoming old-DROPPED/new pair used by alter procedures. + final Set<PipeStaticMeta> replacedLocalMetas = + pipeMetaListFromCoordinator.stream() + .map(meta -> pipeMetaKeeper.getPipeMeta(meta.getStaticMeta())) + .filter(Objects::nonNull) + .map(PipeMeta::getStaticMeta) + .collect(Collectors.toSet()); + final List<PipeMeta> fullPipeMetaList = new ArrayList<>(pipeMetaListFromCoordinator); + for (final PipeMeta localMeta : pipeMetaKeeper.getPipeMetaList()) { + if (!replacedLocalMetas.contains(localMeta.getStaticMeta())) { + fullPipeMetaList.add(localMeta); + } + } + return handlePipeMetaChangesInternal(fullPipeMetaList); + } finally { + releaseWriteLock(); + } + } + @Override public List<TPushPipeMetaRespExceptionMessage> handlePipeMetaChangesInternal( final List<PipeMeta> pipeMetaListFromCoordinator) { 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 1f5f95b4460..509d2904aa6 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 @@ -1451,7 +1451,7 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface final List<TPushPipeMetaRespExceptionMessage> exceptionMessages) { final List<TPushPipeMetaRespExceptionMessage> exceptionMessagesFromAgent = PipeDataNodeAgent.task() - .handlePipeMetaChanges( + .handleMultiPipeMetaChanges( pipeMetas.stream() .map(PipeMeta::deserialize4TaskAgent) .collect(Collectors.toList())); 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 83709d41c3f..7ba4fb7553c 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 @@ -42,7 +42,9 @@ import org.awaitility.Awaitility; import org.junit.Assert; import org.junit.Test; +import java.io.IOException; import java.lang.reflect.Field; +import java.util.Arrays; import java.util.Collections; import java.util.HashMap; import java.util.Map; @@ -55,6 +57,113 @@ public class PipeDataNodeTaskAgentTest { private static final int LOCAL_NODE_ID = 1; private static final int REGION_ID = 7; + @Test + public void testEmptyPartialMetadataKeepsExistingPipe() throws Exception { + final PipeDataNodeTaskAgent agent = newActiveTaskAgent(); + final PipeMetaKeeper keeper = getField(agent, "pipeMetaKeeper"); + final PipeMeta existing = newEmptyPipeMeta("existing", 1L); + addEmptyPipe(agent, existing); + + Assert.assertEquals( + Collections.emptyList(), agent.handleMultiPipeMetaChanges(Collections.emptyList())); + + Assert.assertSame(existing, keeper.getPipeMeta(existing.getStaticMeta())); + } + + @Test + public void testPartialMetadataUpdatesOnlyRequestedPipe() throws Exception { + final PipeDataNodeTaskAgent agent = newActiveTaskAgent(); + final PipeMetaKeeper keeper = getField(agent, "pipeMetaKeeper"); + final PipeMeta existing = newEmptyPipeMeta("existing", 1L); + final PipeMeta unrelated = newEmptyPipeMeta("unrelated", 2L); + addEmptyPipe(agent, existing); + addEmptyPipe(agent, unrelated); + final PipeMeta updated = newEmptyPipeMeta("existing", 1L); + updated.getRuntimeMeta().getStatus().set(PipeStatus.RUNNING); + + Assert.assertEquals( + Collections.emptyList(), + agent.handleMultiPipeMetaChanges(Collections.singletonList(updated))); + + Assert.assertEquals(PipeStatus.RUNNING, existing.getRuntimeMeta().getStatus().get()); + Assert.assertSame(unrelated, keeper.getPipeMeta(unrelated.getStaticMeta())); + Assert.assertEquals(PipeStatus.STOPPED, unrelated.getRuntimeMeta().getStatus().get()); + } + + @Test + public void testPartialDropKeepsUnrelatedPipeAndFullSnapshotStillRemovesIt() throws Exception { + final PipeDataNodeTaskAgent agent = newActiveTaskAgent(); + final PipeMetaKeeper keeper = getField(agent, "pipeMetaKeeper"); + final PipeMeta existing = newEmptyPipeMeta("existing", 1L); + final PipeMeta unrelated = newEmptyPipeMeta("unrelated", 2L); + addEmptyPipe(agent, existing); + addEmptyPipe(agent, unrelated); + final PipeMeta dropped = newEmptyPipeMeta("existing", 1L); + dropped.getRuntimeMeta().getStatus().set(PipeStatus.DROPPED); + + Assert.assertEquals( + Collections.emptyList(), + agent.handleMultiPipeMetaChanges(Collections.singletonList(dropped))); + + Assert.assertNull(keeper.getPipeMeta(existing.getStaticMeta())); + Assert.assertSame(unrelated, keeper.getPipeMeta(unrelated.getStaticMeta())); + Assert.assertTrue(agent.handlePipeMetaChanges(Collections.emptyList()).isEmpty()); + Assert.assertTrue(keeper.isEmpty()); + } + + @Test + public void testPartialAlterKeepsReplacementAndUnrelatedPipe() throws Exception { + final PipeDataNodeTaskAgent agent = newActiveTaskAgent(); + final PipeMetaKeeper keeper = getField(agent, "pipeMetaKeeper"); + final PipeMeta existing = newEmptyPipeMeta("existing", 1L); + final PipeMeta unrelated = newEmptyPipeMeta("unrelated", 2L); + addEmptyPipe(agent, existing); + addEmptyPipe(agent, unrelated); + final PipeMeta dropped = newEmptyPipeMeta("existing", 1L); + dropped.getRuntimeMeta().getStatus().set(PipeStatus.DROPPED); + final PipeMeta replacement = newEmptyPipeMeta("existing", 3L); + + Assert.assertEquals( + Collections.emptyList(), + agent.handleMultiPipeMetaChanges(Arrays.asList(dropped, replacement))); + + Assert.assertEquals( + 3L, keeper.getPipeMeta(replacement.getStaticMeta()).getStaticMeta().getCreationTime()); + Assert.assertSame(unrelated, keeper.getPipeMeta(unrelated.getStaticMeta())); + Assert.assertEquals(2, keeper.getPipeMetaCount()); + } + + private void addEmptyPipe(final PipeDataNodeTaskAgent agent, final PipeMeta meta) + throws Exception { + final PipeMetaKeeper keeper = getField(agent, "pipeMetaKeeper"); + final PipeTaskManager manager = getField(agent, "pipeTaskManager"); + keeper.addPipeMeta(meta); + manager.addPipeTasks(meta.getStaticMeta(), Collections.emptyMap()); + } + + private static PipeDataNodeTaskAgent newActiveTaskAgent() { + return new PipeDataNodeTaskAgent() { + @Override + protected boolean isShutdown() { + return false; + } + }; + } + + private static PipeMeta newEmptyPipeMeta(final String name, final long creationTime) + throws IOException { + return PipeMeta.deserialize4TaskAgent( + new PipeMeta( + new PipeStaticMeta( + name, + creationTime, + Collections.emptyMap(), + Collections.emptyMap(), + Collections.emptyMap()), + new PipeRuntimeMeta()) + .serialize()); + } + @Test public void testGetPipeTaskProgressIndexReportsMissingTaskMeta() throws Exception { final PipeDataNodeTaskAgent taskAgent = new PipeDataNodeTaskAgent();
