Pipe: Optimize realtime performace when pipe starts after long time stop with heavy data backlog (#15048)

[SteveYurongSu]

As title.

[sonarqubecloud]

Quality Gate passed

Issues
5 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

Attention: Patch coverage is 38.70968% with 76 lines in your changes missing coverage. Please review.

Project coverage is 39.39%. Comparing base (af11a09) to head (3a24f70).
Report is 2 commits behind head on master.

Files with missing lines	Patch %	Lines
...otdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java	0.00%	31 Missing ⚠️
...k/connector/PipeRealtimePriorityBlockingQueue.java	0.00%	27 Missing ⚠️
...ol/thrift/async/IoTDBDataRegionAsyncConnector.java	18.18%	9 Missing ⚠️
...g/apache/iotdb/commons/pipe/config/PipeConfig.java	18.18%	9 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #15057      +/-   ##
============================================
+ Coverage     39.37%   39.39%   +0.01%     
  Complexity      193      193              
============================================
  Files          4575     4575              
  Lines        293398   293495      +97     
  Branches      36493    36506      +13     
============================================
+ Hits         115535   115621      +86     
- Misses       177863   177874      +11     

☔ View full report in Codecov by Sentry.
📢 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.