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.24891% with 75 lines in your changes missing coverage. Please review.
✅ Project coverage is 45.93%. Comparing base (5c27559) to head (4a3041f).
⚠️ Report is 1 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 ⚠️
...pe/metric/overview/PipeTsFileToTabletsMetrics.java	54.54%	5 Missing ⚠️
...metric/overview/PipeDataNodeSinglePipeMetrics.java	93.22%	4 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     45.91%   45.93%   +0.02%     
  Complexity      712      712              
============================================
  Files          5502     5502              
  Lines        398764   398872     +108     
  Branches      51963    52002      +39     
============================================
+ Hits         183090   183232     +142     
+ Misses       215674   215640      -34     

☔ 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