Pipe: Fix potential resource leaks in pipe lifecycle cleanup

[Caideyipi]

Description

This PR fixes several potential memory/resource leaks in Pipe runtime lifecycle cleanup.

Previously, some passive paths could create or retain per-schema-region or per-pipe runtime state during leader notification, snapshot handling, queue cleanup, metric updates, pipe drop, and schema-region deletion. This could leave stale listeners, metrics, committers, rate limiters, aggregate runtime state, or floating memory counters after a pipe/schema region was closed, dropped, altered, or deleted.

Main changes:

• Avoid creating schema-region pipe listeners from passive paths.

  • Add schemaListenerIfPresent(...) and use it in snapshot/load/write/delete and queue cleanup paths.
  • Keep schema-region leader-ready state separate from listener state, so leader notifications no longer instantiate SchemaRegionListeningQueue and its metrics.
  • Cleanup unused closed schema listeners and deregister listener metrics.
  • Clear listener and leader-ready state when deleting a schema region.

• Cleanup stale per-pipe runtime state on pipe drop/discard.

  • Clear PipeEventCommitManager committer and restart-time state when a pipe is dropped.
  • Remove endpoint rate limiter entries when data-region sinks discard events of a dropped pipe.
  • Ensure pipe metrics deregistration removes internal maps/gauges/counters, including when the metric service is not bound.
  • Avoid recreating remaining-event metric operators for stale pipe instances during metric updates.

• Make floating memory accounting safe for late event releases.

  • Track floating memory by pipeName_creationTime instead of relying only on live PipeMeta.
  • Keep counters after pipe drop/alter while late releases may still arrive.
  • Cleanup counters only after usage returns to zero and the matching pipe instance is no longer alive.

• Fix aggregate processor shared-state cleanup.

  • Decrease per-pipe reference count safely.
  • Remove and clear per-timeseries runtime state only when the last processor reference closes.

Tests

Added unit tests for:

• PipeSchemaRegionListenerManagerTest: verifies leader state does not create listeners and unused closed listeners are cleaned up.
• PipeTaskAgentTest: verifies floating memory usage remains tracked after pipe drop until late release, and zero counters are cleaned after drop.

---

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

• PipeSchemaRegionListenerManager
• PipeDataNodeRuntimeAgent
• SchemaExecutionVisitor
• SchemaRegionStateMachine
• DataNodeRegionManager
• PipeDataNodeTaskAgent
• PipeEventCommitManager
• IoTDBSink
• PipeDataNodeSinglePipeMetrics
• PipeTsFileToTabletsMetrics
• PipeTaskAgent
• AggregateProcessor

[sonarqubecloud]

Quality Gate passed

Issues
2 New issues
0 Accepted issues

Measures
0 Security Hotspots
0.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud

[codecov]

Codecov Report

❌ Patch coverage is 67.24138% with 76 lines in your changes missing coverage. Please review.
✅ Project coverage is 46.11%. Comparing base (f8135bb) to head (ea2b4e9).
⚠️ Report is 10 commits behind head on master.

Files with missing lines	Patch %	Lines
...machine/schemaregion/SchemaRegionStateMachine.java	10.00%	18 Missing ⚠️
...temachine/schemaregion/SchemaExecutionVisitor.java	0.00%	11 Missing ⚠️
...b/pipe/processor/aggregate/AggregateProcessor.java	0.00%	10 Missing ⚠️
...pe/agent/task/progress/PipeEventCommitManager.java	0.00%	10 Missing ⚠️
...otdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java	0.00%	7 Missing ⚠️
...metric/overview/PipeDataNodeSinglePipeMetrics.java	91.93%	5 Missing ⚠️
...pe/metric/overview/PipeTsFileToTabletsMetrics.java	54.54%	5 Missing ⚠️
...b/pipe/agent/runtime/PipeDataNodeRuntimeAgent.java	50.00%	3 Missing ⚠️
...agent/runtime/PipeSchemaRegionListenerManager.java	90.62%	3 Missing ⚠️
...e/iotdb/commons/pipe/agent/task/PipeTaskAgent.java	96.36%	2 Missing ⚠️
... and 2 more

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #17666      +/-   ##
============================================
+ Coverage     46.03%   46.11%   +0.07%     
  Complexity      712      712              
============================================
  Files          5509     5509              
  Lines        400878   401121     +243     
  Branches      52375    52435      +60     
============================================
+ Hits         184536   184964     +428     
+ Misses       216342   216157     -185     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[sonarqubecloud]

Quality Gate passed

Issues
2 New issues
0 Accepted issues

Measures
0 Security Hotspots
0.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud

[sonarqubecloud]

Quality Gate passed

Issues
21 New issues
0 Accepted issues

Measures
0 Security Hotspots
0.0% Coverage on New Code
2.1% Duplication on New Code

See analysis details on SonarQube Cloud

[hongzhi-gao]

1. iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/progress/PipeEventCommitManager.java:80
   • Trigger: An event with a valid commit ID commits after clear(pipeName, creationTime) finishes during pipe drop.
   • Potential issue: clear removes the committer and restart-time entries, but commitSingleId recreates the restart-time entry with computeIfAbsent when the committer is missing, even when debug logging is disabled. A late commit therefore restores state for the dropped pipe, with no subsequent cleanup shown. The new cleanup does not prevent the lifecycle leak it targets.