This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 5d3e2eb121e Preserve unlisted pipes in partial metadata updates
(#19043)
5d3e2eb121e is described below
commit 5d3e2eb121ed327ca65d89036a7c9c45d4bf8f79
Author: Caideyipi <[email protected]>
AuthorDate: Fri Oct 30 09:49:35 2026 +0800
Preserve unlisted pipes in partial metadata updates (#19043)
* Fix partial pipe metadata reconciliation
* Fix stale partial pipe metadata reconciliation
---
.../db/pipe/agent/task/PipeDataNodeTaskAgent.java | 43 +++++++
.../impl/DataNodeInternalRPCServiceImpl.java | 2 +-
.../pipe/agent/task/PipeDataNodeTaskAgentTest.java | 141 +++++++++++++++++++++
3 files changed, 185 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 5b7775c2315..82e8e2da608 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
@@ -200,6 +200,49 @@ 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 -> {
+ final PipeStaticMeta staticMeta = meta.getStaticMeta();
+ final PipeStatus status =
meta.getRuntimeMeta().getStatus().get();
+ if (status == PipeStatus.DROPPED) {
+ return pipeMetaKeeper.getPipeMeta(
+ staticMeta.getPipeName(),
staticMeta.getCreationTime());
+ }
+ return status == PipeStatus.PRE_DELETE
+ ? null
+ : pipeMetaKeeper.getPipeMeta(staticMeta);
+ })
+ .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
protected TPushPipeMetaRespExceptionMessage
handleSinglePipeMetaChangesInternal(
final PipeMeta pipeMetaFromCoordinator) {
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..fe1bfbe4aab 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,145 @@ 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 testPartialStaleDropDoesNotRemoveNewerPipe() throws Exception {
+ final PipeDataNodeTaskAgent agent = newActiveTaskAgent();
+ final PipeMetaKeeper keeper = getField(agent, "pipeMetaKeeper");
+ final PipeMeta current = newEmptyPipeMeta("existing", 3L);
+ addEmptyPipe(agent, current);
+ final PipeMeta staleDropped = newEmptyPipeMeta("existing", 1L);
+ staleDropped.getRuntimeMeta().getStatus().set(PipeStatus.DROPPED);
+
+ Assert.assertEquals(
+ Collections.emptyList(),
+
agent.handleMultiPipeMetaChanges(Collections.singletonList(staleDropped)));
+
+ Assert.assertSame(current, keeper.getPipeMeta(current.getStaticMeta()));
+ }
+
+ @Test
+ public void testPartialPreDeleteDoesNotRemoveExistingPipe() throws Exception
{
+ final PipeDataNodeTaskAgent agent = newActiveTaskAgent();
+ final PipeMetaKeeper keeper = getField(agent, "pipeMetaKeeper");
+ final PipeMeta current = newEmptyPipeMeta("existing", 3L);
+ addEmptyPipe(agent, current);
+ final PipeMeta preDelete = newEmptyPipeMeta("existing", 4L);
+ preDelete.getRuntimeMeta().getStatus().set(PipeStatus.PRE_DELETE);
+
+ Assert.assertEquals(
+ Collections.emptyList(),
+
agent.handleMultiPipeMetaChanges(Collections.singletonList(preDelete)));
+
+ Assert.assertSame(current, keeper.getPipeMeta(current.getStaticMeta()));
+ }
+
+ @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();
