[ISSUE #11089] Share consumption executors for Proxy internal clients

[qianye1001]

Which Issue(s) This PR Fixes

Fixes #11089

Brief Description

Proxy internal system-message PushConsumers currently create dedicated consumption pools. Allow them to share a Proxy-owned executor to reduce consumption threads while preserving queued messages.

DefaultMQPushConsumer.setConsumeExecutor(ExecutorService) supplies an externally managed executor directly to pull/POP consumption with concurrent/orderly listeners, including virtual-thread executors supplied on JDK 21+. A shared AbstractConsumeMessageService centralizes executor construction, ownership, shutdown and core-pool controls. Consumers shut down and resize only their own pools; the caller owns an injected executor's lifecycle. Without injection, existing executor construction and consumer behavior remain unchanged.

Proxy configures the shared pool through systemMessageConsumerThreadPoolCoreSize, defaulting to twice the available processors. Core and maximum sizes both use this value. The pool uses an unbounded LinkedBlockingQueue and AbortPolicy, with core-thread timeout disabled. Queued consumption tasks are retained rather than evicted, and executor shutdown remains observable through rejection.

The setter Javadoc documents external ownership and warns that silently discarding tasks or merely cancelling their futures can retain ProcessQueue messages and pin offsets. Consumer groups, subscriptions and offset storage are unchanged.

How Did You Test This Change?

• JDK 11: the shared executor, consumer sharing and HeartbeatSyncer regression suites passed: 7 tests, no failures or skips.
• Checkstyle and SpotBugs passed across the Proxy reactor build.
• The queue regression test blocks the worker, queues 10,001 tasks and verifies that every task completes without eviction, followed by observable rejection after shutdown.
• Existing tests cover executor identity, caller-owned shutdown, no resizing of shared pools, ordinary-consumer offset preservation on cancellation and constructor wiring.

[RockteMQ-AI]

Summary

This PR introduces executor injection for PushConsumers (setConsumeExecutor) and a shared SystemMessageConsumeExecutor in Proxy, reducing thread overhead for system-message consumers. A new AbstractConsumeMessageService cleanly centralizes executor ownership, lifecycle, and pool-size controls across all four consume services.

Overall the design is solid — ownership semantics are explicit (ownsConsumeExecutor), backward compatibility is preserved (null external executor → default dedicated pool), and the Proxy discard-oldest rejection policy correctly delegates to consumeFailed() for broadcast offset cleanup. Test coverage is comprehensive, including virtual-thread scenarios on JDK 21.

Suggestions

• [Info] getCorePoolSize() now returns -1 for externally-managed executors. Callers that cache or display this value (e.g., monitoring dashboards, admin tools) should be aware of the sentinel. Consider adding a brief note in the PR description or release notes.

• [Info] systemMessageConsumerThreadPoolQueueCapacity defaults to 10,000. Under bursty heartbeat traffic this could buffer a large number of tasks before rejection kicks in. The default seems reasonable for normal operation, but operators should be aware that DiscardOldestPolicy will silently drop the oldest buffered task when full (with the cleanup path you implemented).

• [Info] The AbstractConsumeMessageService constructor always creates a ThreadFactoryImpl(ConsumeMessageThread_) even when an external executor is injected (where it goes unused). This is harmless but could be made lazy or conditional for clarity.

No blocking issues found. LGTM.

---

Automated review by github-manager-bot

[codecov-commenter]

Codecov Report

❌ Patch coverage is 82.14286% with 10 lines in your changes missing coverage. Please review.
✅ Project coverage is 48.89%. Comparing base (4fd0e3b) to head (b3c92c9).

Files with missing lines	Patch %	Lines
.../rocketmq/proxy/service/ClusterServiceManager.java	0.00%	3 Missing ⚠️
...ervice/sysmessage/AbstractSystemMessageSyncer.java	40.00%	3 Missing ⚠️
...t/impl/consumer/AbstractConsumeMessageService.java	90.47%	0 Missing and 2 partials ⚠️
...q/proxy/service/client/ClusterConsumerManager.java	33.33%	2 Missing ⚠️

Additional details and impacted files

@@              Coverage Diff              @@
##             develop   #11090      +/-   ##
=============================================
- Coverage      48.93%   48.89%   -0.05%     
+ Complexity     13846    13839       -7     
=============================================
  Files           1382     1384       +2     
  Lines         101602   101596       -6     
  Branches       13213    13210       -3     
=============================================
- Hits           49717    49672      -45     
- Misses         45858    45882      +24     
- Partials        6027     6042      +15     

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

[RockteMQ-AI]

Summary

The new commits simplify the shared executor design by switching to an unbounded queue with AbortPolicy, removing the complex discard/cleanup logic. This eliminates the risk of silent message loss and reduces code complexity. The increased default core pool size and removal of core thread timeout are appropriate for long-running Proxy workloads.

Changes reviewed:

• Commit 9aa51ab: Preserved queued system messages by removing DiscardOldestPolicy and custom cleanup
• Commit b3c92c9: Kept shared consumption core threads alive (removed allowCoreThreadTimeOut)

LGTM — the simplification is well-reasoned and the code is clean.

---

Automated review by RockteMQ-AI