This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new 505b18a8af3 fix(pipe): reuse safe local progress on alter (#18423)
(#18444)
505b18a8af3 is described below
commit 505b18a8af3667fbac6df0c232e0500a7ea42bfd
Author: Caideyipi <[email protected]>
AuthorDate: Wed Aug 12 09:35:35 2026 +0800
fix(pipe): reuse safe local progress on alter (#18423) (#18444)
* fix(pipe): reuse safe local progress on alter
* fix(pipe): collect latest progress before alter
---
.../iotdb/pipe/it/autocreate/IoTDBPipeAlterIT.java | 48 +++++++
.../procedure/env/ConfigNodeProcedureEnv.java | 33 +++++
.../impl/pipe/task/AlterPipeProcedureV2.java | 83 ++++++++++-
.../db/pipe/agent/task/PipeDataNodeTaskAgent.java | 83 +++++++++++
.../impl/DataNodeInternalRPCServiceImpl.java | 18 +++
.../thrift/impl/PushMultiPipeMetaHelper.java | 36 ++++-
.../pipe/agent/task/PipeDataNodeTaskAgentTest.java | 159 +++++++++++++++++++++
...nternalRPCServiceImplPushMultiPipeMetaTest.java | 72 ++++++++++
8 files changed, 524 insertions(+), 8 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeAlterIT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeAlterIT.java
index b9644159386..9605c4874a4 100644
---
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeAlterIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeAlterIT.java
@@ -47,6 +47,15 @@ import static org.junit.Assert.fail;
@Category({MultiClusterIT2AutoCreateSchema.class})
public class IoTDBPipeAlterIT extends AbstractPipeDualAutoIT {
+ @Override
+ protected void setupConfig() {
+ super.setupConfig();
+ senderEnv
+ .getConfig()
+ .getCommonConfig()
+ .setPipeHeartbeatIntervalSecondsForCollectingPipeMeta(600);
+ }
+
@Test
public void testBasicAlterPipe() throws Exception {
final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
@@ -579,4 +588,43 @@ public class IoTDBPipeAlterIT extends
AbstractPipeDualAutoIT {
"count(timeseries),",
Collections.singleton("1,"));
}
+
+ @Test
+ public void testAlterPipeDoesNotResendCommittedData() {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ TestUtils.executeNonQueries(
+ senderEnv,
+ Arrays.asList("insert into root.db.d1(time, s1) values (1, 1), (2,
2)", "flush"),
+ null);
+
+ TestUtils.executeNonQuery(
+ senderEnv,
+ String.format(
+ "create pipe a2b with source ('source.realtime.mode'='stream')
with sink ('node-urls'='%s', 'sink.batch.enable'='false')",
+ receiverDataNode.getIpAndPortString()),
+ null);
+
+ final Set<String> oldData = new HashSet<>(Arrays.asList("1,1.0,",
"2,2.0,"));
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv, "select * from root.db.d1", "Time,root.db.d1.s1,",
oldData);
+
+ TestUtils.executeNonQuery(
+ receiverEnv, "delete from root.db.d1.s1 where time >= 1 and time <=
2", null);
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv, "select * from root.db.d1", "Time,root.db.d1.s1,",
Collections.emptySet());
+
+ TestUtils.executeNonQuery(
+ senderEnv, "alter pipe a2b modify sink ('sink.batch.enable'='true')",
null);
+ TestUtils.assertDataAlwaysOnEnv(
+ receiverEnv, "select * from root.db.d1", "Time,root.db.d1.s1,",
Collections.emptySet());
+
+ TestUtils.executeNonQueries(
+ senderEnv, Arrays.asList("insert into root.db.d1(time, s1) values (3,
3)", "flush"), null);
+ final Set<String> newData = Collections.singleton("3,3.0,");
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv, "select * from root.db.d1", "Time,root.db.d1.s1,",
newData);
+ TestUtils.assertDataAlwaysOnEnv(
+ receiverEnv, "select * from root.db.d1", "Time,root.db.d1.s1,",
newData);
+ }
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java
index 174bc2ca6dc..ae7fedd2c50 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java
@@ -24,6 +24,7 @@ import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
import org.apache.iotdb.common.rpc.thrift.TDataNodeConfiguration;
import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
+import org.apache.iotdb.common.rpc.thrift.TPipeHeartbeatResp;
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.cluster.NodeStatus;
@@ -73,6 +74,7 @@ import
org.apache.iotdb.mpp.rpc.thrift.TDropTriggerInstanceReq;
import org.apache.iotdb.mpp.rpc.thrift.TInactiveTriggerInstanceReq;
import org.apache.iotdb.mpp.rpc.thrift.TInvalidateCacheReq;
import org.apache.iotdb.mpp.rpc.thrift.TNotifyRegionMigrationReq;
+import org.apache.iotdb.mpp.rpc.thrift.TPipeHeartbeatReq;
import org.apache.iotdb.mpp.rpc.thrift.TPushConsumerGroupMetaReq;
import org.apache.iotdb.mpp.rpc.thrift.TPushConsumerGroupMetaResp;
import org.apache.iotdb.mpp.rpc.thrift.TPushMultiPipeMetaReq;
@@ -953,6 +955,37 @@ public class ConfigNodeProcedureEnv {
.collect(Collectors.toList());
}
+ /**
+ * Collect the current pipe metadata from the specified DataNodes before a
metadata-changing
+ * procedure. The caller can use the returned task progress to avoid basing
a replacement pipe on
+ * a stale ConfigNode heartbeat.
+ *
+ * <p>This is deliberately best effort. A DataNode that is unavailable
cannot contribute a newer
+ * checkpoint, but the alter procedure can still use the checkpoint already
stored by ConfigNode.
+ */
+ public Map<Integer, TPipeHeartbeatResp> collectPipeMetaFromDataNodes(
+ final Set<Integer> dataNodeIds) {
+ if (dataNodeIds.isEmpty()) {
+ return Collections.emptyMap();
+ }
+
+ final Map<Integer, TDataNodeLocation> dataNodeLocationMap =
+
configManager.getNodeManager().getRegisteredDataNodeLocations().entrySet().stream()
+ .filter(entry -> dataNodeIds.contains(entry.getKey()))
+ .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+ if (dataNodeLocationMap.isEmpty()) {
+ return Collections.emptyMap();
+ }
+
+ final DataNodeAsyncRequestContext<TPipeHeartbeatReq, TPipeHeartbeatResp>
clientHandler =
+ new DataNodeAsyncRequestContext<>(
+ CnToDnAsyncRequestType.PIPE_HEARTBEAT,
+ new TPipeHeartbeatReq(System.currentTimeMillis()),
+ dataNodeLocationMap);
+ sendRuntimeMetaRequest(clientHandler, true,
getRuntimeMetaPushTimeoutInMs());
+ return clientHandler.getResponseMap();
+ }
+
private static Map<Integer, TPushPipeMetaResp> sendPipeMetaRequest(
final DataNodeAsyncRequestContext<?, TPushPipeMetaResp> clientHandler,
final long timeoutInMs,
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/AlterPipeProcedureV2.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/AlterPipeProcedureV2.java
index dcebb0b3d33..f7ed7496f77 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/AlterPipeProcedureV2.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/AlterPipeProcedureV2.java
@@ -20,7 +20,9 @@
package org.apache.iotdb.confignode.procedure.impl.pipe.task;
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
+import org.apache.iotdb.common.rpc.thrift.TPipeHeartbeatResp;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.consensus.index.ProgressIndex;
import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
import org.apache.iotdb.commons.pipe.agent.task.PipeTaskAgent;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta;
@@ -33,6 +35,7 @@ import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
import
org.apache.iotdb.confignode.consensus.request.write.pipe.task.AlterPipePlanV2;
import org.apache.iotdb.confignode.manager.pipe.coordinator.PipeManager;
+import
org.apache.iotdb.confignode.manager.pipe.coordinator.runtime.heartbeat.PipeHeartbeat;
import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
import
org.apache.iotdb.confignode.procedure.impl.pipe.AbstractOperatePipeProcedureV2;
import org.apache.iotdb.confignode.procedure.impl.pipe.PipeTaskOperation;
@@ -49,9 +52,12 @@ import org.slf4j.LoggerFactory;
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.Map;
import java.util.Objects;
+import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@@ -138,6 +144,16 @@ public class AlterPipeProcedureV2 extends
AbstractOperatePipeProcedureV2 {
new HashMap<>(alterPipeRequest.getProcessorAttributes()),
new HashMap<>(alterPipeRequest.getConnectorAttributes()));
+ // The periodic heartbeat may not have reached ConfigNode immediately
before this alter. Pull
+ // the current leader checkpoints first so a leader migration does not
make the replacement
+ // task start from an older coordinator checkpoint.
+ final Map<Integer, ProgressIndex> latestProgressIndexMap =
+
PipeTaskAgent.isRealtimeOnlyPipe(currentPipeStaticMeta.getExtractorParameters())
+ ==
PipeTaskAgent.isRealtimeOnlyPipe(updatedPipeStaticMeta.getExtractorParameters())
+ ? collectLatestProgressIndexes(
+ env, currentPipeStaticMeta,
currentConsensusGroupId2PipeTaskMeta)
+ : Collections.emptyMap();
+
final ConcurrentMap<Integer, PipeTaskMeta>
updatedConsensusGroupIdToTaskMetaMap =
new ConcurrentHashMap<>();
@@ -167,8 +183,7 @@ public class AlterPipeProcedureV2 extends
AbstractOperatePipeProcedureV2 {
// then it will extract all existing data now, not existing
data since the
// original pipe was created
// Similar for "pure realtime"
- updatedConsensusGroupIdToTaskMetaMap.put(
- regionGroupId.getId(),
+ final PipeTaskMeta updatedPipeTaskMeta =
new PipeTaskMeta(
PipeTaskAgent.isRealtimeOnlyPipe(
currentPipeStaticMeta.getExtractorParameters())
@@ -186,7 +201,14 @@ public class AlterPipeProcedureV2 extends
AbstractOperatePipeProcedureV2 {
&& PipeTaskAgent.isRealtimeOnlyPipe(
updatedPipeStaticMeta.getExtractorParameters()))
?
PipeTaskMeta.getRevertedLeader(regionLeaderNodeId)
- : regionLeaderNodeId));
+ : regionLeaderNodeId);
+ final ProgressIndex latestProgressIndex =
+ latestProgressIndexMap.get(regionGroupId.getId());
+ if (latestProgressIndex != null) {
+ updatedPipeTaskMeta.updateProgressIndex(latestProgressIndex);
+ }
+ updatedConsensusGroupIdToTaskMetaMap.put(
+ regionGroupId.getId(), updatedPipeTaskMeta);
}
});
@@ -213,6 +235,61 @@ public class AlterPipeProcedureV2 extends
AbstractOperatePipeProcedureV2 {
}
}
+ private Map<Integer, ProgressIndex> collectLatestProgressIndexes(
+ final ConfigNodeProcedureEnv env,
+ final PipeStaticMeta pipeStaticMeta,
+ final Map<Integer, PipeTaskMeta> taskMetaMap) {
+ final Set<Integer> leaderNodeIds = new HashSet<>();
+ final Set<Integer> registeredDataNodeIds =
+
env.getConfigManager().getNodeManager().getRegisteredDataNodeLocations().keySet();
+ taskMetaMap.forEach(
+ (consensusGroupId, taskMeta) -> {
+ // The ConfigRegion task is led by a ConfigNode, not by a DataNode.
+ if (consensusGroupId != Integer.MIN_VALUE
+ && registeredDataNodeIds.contains(taskMeta.getLeaderNodeId())) {
+ leaderNodeIds.add(taskMeta.getLeaderNodeId());
+ }
+ });
+
+ if (leaderNodeIds.isEmpty()) {
+ return Collections.emptyMap();
+ }
+
+ final Map<Integer, TPipeHeartbeatResp> responseMap =
+ env.collectPipeMetaFromDataNodes(leaderNodeIds);
+ final Map<Integer, ProgressIndex> latestProgressIndexMap = new HashMap<>();
+ responseMap.forEach(
+ (dataNodeId, response) -> {
+ if (response == null || !response.isSetPipeMetaList()) {
+ return;
+ }
+
+ final PipeMeta pipeMetaFromDataNode =
+ new PipeHeartbeat(response.getPipeMetaList(), null, null, null)
+ .getPipeMeta(pipeStaticMeta);
+ if (pipeMetaFromDataNode == null) {
+ return;
+ }
+
+ pipeMetaFromDataNode
+ .getRuntimeMeta()
+ .getConsensusGroupId2TaskMetaMap()
+ .forEach(
+ (consensusGroupId, taskMetaFromDataNode) -> {
+ final PipeTaskMeta taskMetaFromCoordinator =
taskMetaMap.get(consensusGroupId);
+ if (taskMetaFromCoordinator == null
+ || taskMetaFromCoordinator.getLeaderNodeId() !=
dataNodeId) {
+ return;
+ }
+ latestProgressIndexMap.merge(
+ consensusGroupId,
+ taskMetaFromDataNode.getProgressIndex(),
+
ProgressIndex::updateToMinimumEqualOrIsAfterProgressIndex);
+ });
+ });
+ return latestProgressIndexMap;
+ }
+
@Override
public void executeFromWriteConfigNodeConsensus(ConfigNodeProcedureEnv env)
throws PipeException {
LOGGER.info(
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 a28cdb95211..d99d5b3cf17 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
@@ -101,6 +101,7 @@ import java.util.concurrent.Future;
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;
import java.util.stream.StreamSupport;
@@ -198,6 +199,8 @@ public class PipeDataNodeTaskAgent extends PipeTaskAgent {
return Collections.emptyList();
}
+ carryOverLocalProgressIndexForAlter(pipeMetaListFromCoordinator);
+
final List<TPushPipeMetaRespExceptionMessage> exceptionMessages =
super.handlePipeMetaChangesInternal(pipeMetaListFromCoordinator);
@@ -218,6 +221,86 @@ public class PipeDataNodeTaskAgent extends PipeTaskAgent {
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.
+ *
+ * <p>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<PipeMeta> 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.getPipeName());
+ if (localOldPipeMeta == null) {
+ continue;
+ }
+
+ for (final PipeMeta updatedPipeMeta : pipeMetaListFromCoordinator) {
+ if (updatedPipeMeta == droppedPipeMeta
+ || updatedPipeMeta.getRuntimeMeta().getStatus().get() ==
PipeStatus.DROPPED
+ ||
!oldStaticMeta.getPipeName().equals(updatedPipeMeta.getStaticMeta().getPipeName()))
{
+ 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<PipeStaticMeta, Integer> 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.getExtractorParameters())
+ !=
PipeTaskAgent.isRealtimeOnlyPipe(updatedStaticMeta.getExtractorParameters())) {
+ return;
+ }
+
+ final Map<Integer, PipeTaskMeta> localTaskMetaMap =
+ localOldPipeMeta.getRuntimeMeta().getConsensusGroupId2TaskMetaMap();
+ final Map<Integer, PipeTaskMeta> updatedTaskMetaMap =
+ updatedPipeMeta.getRuntimeMeta().getConsensusGroupId2TaskMetaMap();
+
+ for (final Map.Entry<Integer, PipeTaskMeta> 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<Integer> clearSchemaRegionListeningQueueIfNecessary(
final List<PipeMeta> pipeMetaListFromCoordinator) throws
IllegalPathException {
final Map<Integer, Long> 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 1d974146464..99e314affa0 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
@@ -1163,6 +1163,24 @@ public class DataNodeInternalRPCServiceImpl implements
IDataNodeRPCService.Iface
return PipeDataNodeAgent.task().handleDropPipe(pipeName);
}
+ @Override
+ public boolean handlePipeMetaChanges(
+ final List<ByteBuffer> pipeMetas,
+ final List<TPushPipeMetaRespExceptionMessage> exceptionMessages)
{
+ final List<TPushPipeMetaRespExceptionMessage>
exceptionMessagesFromAgent =
+ PipeDataNodeAgent.task()
+ .handlePipeMetaChanges(
+ pipeMetas.stream()
+ .map(PipeMeta::deserialize4TaskAgent)
+ .collect(Collectors.toList()));
+ // PipeTaskAgent returns null only when its timed write-lock
acquisition fails.
+ 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 73fdbf804de..f6a82603065 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
@@ -45,6 +45,32 @@ final class PushMultiPipeMetaHelper {
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.
+ *
+ * <p>The default implementation preserves the per-metadata behavior for
handlers that do not
+ * need batch processing.
+ *
+ * @param pipeMetas serialized pipe metadata to process as one batch
+ * @param exceptionMessages destination for per-pipe failures
+ * @return {@code true} after the batch was processed, or {@code false}
only when the task agent
+ * could not acquire its write lock before the metadata handling
timeout
+ */
+ default boolean handlePipeMetaChanges(
+ final List<ByteBuffer> pipeMetas,
+ final List<TPushPipeMetaRespExceptionMessage> exceptionMessages)
+ throws Exception {
+ for (final ByteBuffer pipeMeta : pipeMetas) {
+ final TPushPipeMetaRespExceptionMessage message =
handleSinglePipeMeta(pipeMeta);
+ if (message != null) {
+ exceptionMessages.add(message);
+ }
+ }
+ return true;
+ }
}
static TPushPipeMetaResp pushMultiPipeMeta(
@@ -59,11 +85,11 @@ final class PushMultiPipeMetaHelper {
}
}
} else if (req.isSetPipeMetas()) {
- for (final ByteBuffer pipeMeta : req.getPipeMetas()) {
- final TPushPipeMetaRespExceptionMessage message =
handler.handleSinglePipeMeta(pipeMeta);
- if (message != null) {
- exceptionMessages.add(message);
- }
+ // A false result is reserved for the task-agent write-lock timeout.
Per-pipe processing
+ // failures are returned through exceptionMessages and use
PIPE_PUSH_META_ERROR below.
+ if (!handler.handlePipeMetaChanges(req.getPipeMetas(),
exceptionMessages)) {
+ return new TPushPipeMetaResp()
+ .setStatus(new
TSStatus(TSStatusCode.PIPE_PUSH_META_TIMEOUT.getStatusCode()));
}
} else {
throw new Exception("Invalid TPushMultiPipeMetaReq");
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 d3249933aa1..3dd93e87789 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.Assert;
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 class PipeDataNodeTaskAgentTest {
.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<String, String> 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<Integer, PipeTaskMeta> 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 0ae284bce9d..8dc6523d8ed 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 org.junit.Test;
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 class
DataNodeInternalRPCServiceImplPushMultiPipeMetaTest {
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<ByteBuffer> pipeMetas,
+ final List<TPushPipeMetaRespExceptionMessage>
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<ByteBuffer> pipeMetas,
+ final List<TPushPipeMetaRespExceptionMessage>
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);
}
