Caideyipi commented on code in PR #18512:
URL: https://github.com/apache/iotdb/pull/18512#discussion_r3842318547
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java:
##########
@@ -584,6 +582,58 @@ private PipeMetaReport collectPipeMetaReport(
return report;
}
+ // Returns whether every expected DataRegion has a completed local PipeTask
on this DataNode.
+ // An empty expected set means this DataNode does not need to transfer
history and is completed.
+ // A missing PipeTaskMap or a missing expected DataRegion means
initialization failed.
+ static boolean isAllExpectedDataRegionCompleted(
+ final Map<Integer, PipeTask> pipeTaskMap, final Set<Integer>
expectedDataRegionIds) {
+ if (expectedDataRegionIds.isEmpty()) {
+ // This DataNode does not own any target DataRegion for the pipe, so
there is no local
+ // history transfer to wait for.
+ return true;
+ }
+ return pipeTaskMap != null
+ && expectedDataRegionIds.stream()
+ .allMatch(
+ dataRegionId -> {
+ final PipeTask pipeTask = pipeTaskMap.get(dataRegionId);
+ return pipeTask instanceof PipeDataNodeTask
+ && ((PipeDataNodeTask) pipeTask).isCompleted();
+ });
+ }
+
+ // Returns the DataRegion ids that this DataNode is expected to transfer for
the given pipe.
+ // A region is included only when it is owned by this DataNode, is led by
this DataNode according
+ // to the pipe's runtime metadata, and is selected by the pipe's source
parameters. This expected
+ // set is used instead of the already-created PipeTask map so that a failed
task initialization is
+ // not silently treated as a completed region.
+ private Set<Integer> getExpectedDataRegionIds(final PipeMeta pipeMeta) {
+ final PipeStaticMeta staticMeta = pipeMeta.getStaticMeta();
+ final PipeParameters sourceParameters = staticMeta.getSourceParameters();
+ final Set<Integer> localDataRegionIds =
+ StorageEngine.getInstance().getAllDataRegionIds().stream()
+ .map(DataRegionId::getId)
+ .collect(Collectors.toSet());
+ final Set<Integer> expectedDataRegionIds = new HashSet<>();
+ for (final Map.Entry<Integer, PipeTaskMeta> entry :
+
pipeMeta.getRuntimeMeta().getConsensusGroupId2TaskMetaMap().entrySet()) {
Review Comment:
**[P1] Keep failed assignments in the expected set**
`pipeMeta` here is the DataNode's locally accepted runtime meta, not the
coordinator's authoritative meta. During a leader change,
`executeSinglePipeRuntimeMetaChanges` first calls `dropPipeTask`, which removes
the region from this map, and `PipeDataNodeTaskAgent.createPipeTask` puts it
back only after the new connector/task has been created. If connector
initialization throws, the entry remains absent. On a node whose only target is
that region, this method therefore returns an empty set, line 590 treats it as
complete, and ConfigNode can auto-drop the incomplete snapshot—the failure this
PR is intended to prevent. Please retain the coordinator-assigned region ID
even when task creation fails (and treat an assigned-but-not-loaded local
region as incomplete), and cover this failed leader-change/task-creation path.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
