Fix snapshot pipe auto-drop when some DataNodes fail to initialize

[luoluoyuyu]

Problem

Snapshot/backup pipes can be automatically dropped even when some DataNodes failed to initialize their PipeConnector.

In the failing case, ConfigNode reported PIPE_PUSH_SINGLE_META failures with TSStatus(code:1807) and Broken pipe on multiple DataNodes, then PIPE_PUSH_ALL_META timed out. Some DataNodes therefore had no local PipeTask, but the old DataNode completion check treated a missing task map as completed. ConfigNode then saw every DataNode as completed and removed the pipe, making an incomplete backup look successful.

Root cause

DataNode treated pipeTaskMap == null as all regions completed.

DataNode only checked existing tasks, so a Region whose PipeTask failed to start was silently ignored.

ConfigNode waited for all registered DataNodes instead of only the DataNodes that actually own target Regions for the pipe.

ConfigNode subtracted completed nodes from a live keySet() view, which could mutate the registered DataNode set.

Changes

DataNode completion checkCompute the expected local DataRegion IDs from the pipe's runtime metadata. A DataNode is only reported completed when every expected local DataRegion has a completed PipeDataNodeTask. An empty expected set is treated as completed because that DataNode has no local history transfer to perform.

ConfigNode completion checkDerive the expected DataNode IDs from the pipe's runtime metadata instead of using all registered DataNodes. DataNodes that do not own any target Region are ignored during the auto-drop judgment. DataNodes that should participate but have not completed still block auto-drop.

Avoid mutating the registered DataNode setCopy the registered DataNode IDs before subtracting completed IDs, preventing removeAll from modifying the live keySet().

Add unit testAdd coverage for the DataNode completion predicate, including a missing expected Region, a missing task map, an uncompleted task, and an empty expected set.

---

This PR has:

• been self-reviewed.
  • concurrent read
  • concurrent write
  • concurrent read and write
• added documentation for new or modified features or behaviors.
• added Javadocs for most classes and all non-trivial methods.
• added or updated version, license, or notice information
• added comments explaining the "why" and the intent of the code wherever would not be obvious
  for an unfamiliar reader.
• added unit tests or modified existing tests to cover new code paths, ensuring the threshold
  for code coverage.
• added integration tests.
• been tested in a test IoTDB cluster.

---

Key changed/added classes (or packages if there are too many classes) in this PR

[Caideyipi]

[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.