[ISSUE #10906] Prevent duplicate MQClientInstance creation

[qianye1001]

Which Issue(s) This PR Fixes

• Fixes [Bug] Concurrent MQClientInstance creation leaks scheduler threads and periodic tasks #10906

Brief Description

Concurrent first access to MQClientManager previously used get -> construct -> putIfAbsent. Callers racing on one client ID could therefore construct multiple MQClientInstance candidates. Every candidate starts MQClientFactoryScheduledThread during ConsumerStatsManager construction and registers 30 periodic StatsItemSet tasks, so discarded candidates leaked live scheduler threads and tasks.

The unmodified-code reproduction used 8 synchronized callers and observed:

candidates=8
returnedIdentity=1
liveSchedulerThreads=8
periodicTasks=240
losersGc=7/7

The losing outer instances were garbage-collected while the scheduler threads and periodic tasks remained live, confirming the resource reference chain independently of MQClientInstance reachability.

This change:

• uses ConcurrentHashMap.computeIfAbsent so one client ID has at most one successful construction and all callers receive the same instance;
• clones ClientConfig before entering the mapping function, keeping overridable configuration code outside the map's per-bin reservation;
• does not add a global lock, so independent client IDs in different map bins can construct concurrently;
• rolls back the stats scheduler, optional heartbeat executor, and remoting client if construction fails after resources have been allocated;
• rolls back remoting, scheduled tasks, pull, rebalance, inner-producer, and heartbeat resources if start() fails, removes the failed mapping, and preserves the existing START_FAILED state;
• adds identity-aware map removal so a stale instance's delayed shutdown cannot remove a newer replacement.

Design trade-offs

• A global synchronized section would unnecessarily serialize every client ID.
• A removable per-key lock map has waiter/lifecycle and ABA hazards; retaining locks forever would introduce a different leak.
• Moving all constructor resources to start() would be a much larger lifecycle and compatibility change, and by itself would not prevent duplicate candidates.
• computeIfAbsent provides the required atomic publication and exception behavior with the existing Java 8 ConcurrentHashMap: an exception installs no mapping, and a later call may retry.

The fast-path get and existing public removal method remain intact. Normal repeated start()/shutdown() behavior and the START_FAILED retry rejection on the same object are preserved.

Resource audit

The constructor/start path was audited for thread, scheduled-task, and retained-resource creation:

• ConsumerStatsManager: five StatsItemSet objects with six periodic tasks each; zero-delay sampling starts MQClientFactoryScheduledThread during construction.
• MQClientAPIImpl -> NettyRemotingClient: timer, event-loop/selector, public and scan executors; most threads are lazy, but the remoting client is shut down on partial construction/start failure.
• PullMessageService: service thread plus its scheduled executor; both are stopped on start failure.
• RebalanceService: service thread stopped on start failure.
• Inner DefaultMQProducer: registration/executors and request-future/detector lifecycle are shut down when started.
• Optional concurrent-heartbeat executor: shut down on construction or start failure.
• MQClientInstance.startScheduledTask(): all client periodic tasks use the scheduler that is shut down on failure.

How Did You Test This Change?

All commands used the repository-configured Maven source/target 1.8; no compiler target was changed.

1. Full client reactor regression on Amazon Corretto 11.0.23:

   mvn -pl client -am -DskipITs -Dspotbugs.skip=true test

   Results:

   • rocketmq-common: 243 tests, 0 failures/errors
   • rocketmq-remoting: 174 tests, 0 failures/errors
   • rocketmq-client: 585 tests, 0 failures/errors, 1 skipped
   • Reactor: BUILD SUCCESS (3:45)

2. Focused regression plus SpotBugs/checkstyle on Amazon Corretto 11.0.23:

   mvn -pl client -am -DskipITs \
     -Dtest=MQClientManagerTest \
     -Dsurefire.failIfNoSpecifiedTests=false test

   Results: 7 tests passed; SpotBugs reported 0 findings and checkstyle reported 0 violations.

3. Java 8 compatibility on Oracle JDK 8u291:

   mvn -pl client -am -DskipITs -Dspotbugs.skip=true \
     -Dtest=MQClientManagerTest \
     -Dsurefire.failIfNoSpecifiedTests=false test

   Results: 7 tests passed. javap -verbose reports class-file major version 52.

The new tests use barriers, bounded futures, Awaitility assertions, executor termination waits, and isolated thread groups rather than fixed sleeps. They cover:

• high-concurrency same-client-ID creation: one instance, one scheduler, no orphan candidates;
• concurrent construction for different client IDs;
• late constructor failure: no map entry or scheduler thread, followed by successful retry;
• removal and recreation;
• repeated start/shutdown and recreation;
• mid-start failure rollback and replacement creation;
• stale shutdown preserving a newer replacement.

Compatibility / release note

Before this change, an MQClientInstance that failed during start() remained in MQClientManager, so later lookups for the same client ID returned the same START_FAILED instance. After this change, the failed instance is removed by identity after best-effort cleanup, and a later getOrCreateMQClientInstance call may create a replacement.

Callers that already hold the failed instance still observe START_FAILED and are not migrated automatically. If a persistent configuration or environment error is retried without backoff, callers may repeatedly create and clean up replacement instances; retry and backoff policy remains the caller responsibility.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 77.27273% with 20 lines in your changes missing coverage. Please review.
✅ Project coverage is 48.33%. Comparing base (e345861) to head (1385ac7).

Files with missing lines	Patch %	Lines
...rocketmq/client/impl/factory/MQClientInstance.java	75.00%	16 Missing and 4 partials ⚠️

Additional details and impacted files

@@              Coverage Diff              @@
##             develop   #10907      +/-   ##
=============================================
- Coverage      48.44%   48.33%   -0.11%     
+ Complexity     13576    13550      -26     
=============================================
  Files           1380     1380              
  Lines         101165   101195      +30     
  Branches       13127    13130       +3     
=============================================
- Hits           49009    48916      -93     
- Misses         46196    46293      +97     
- Partials        5960     5986      +26     

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

• Identity-aware removal (ConcurrentMap.remove(key, value)) in shutdown() and cleanupAfterStartFailure() correctly prevents ABA issues
• cleanupAfterConstructionFailure properly handles the case where clientAPI is null (failure before MQClientAPIImpl creation)
• runCleanup correctly suppresses secondary exceptions without duplicating the primary cause

Performance ✅

• No global lock added — independent client IDs in different ConcurrentHashMap bins can still construct concurrently
• The bin lock is held during construction, but this is acceptable since construction is infrequent relative to lookups (the fast-path get() is lock-free)

Tests ✅

• 433 lines of comprehensive tests covering concurrent creation, resource cleanup on failure, identity-aware removal, and GC behavior
• Tests verify the exact bug scenario (8 concurrent callers, only 1 instance created, no leaked threads)

Compatibility ✅

• Existing API preserved — the new removeClientFactory(String, MQClientInstance) is additive
• No behavioral change for single-threaded callers

Minor Note

The factoryIndexGenerator.getAndIncrement() inside the lambda is only consumed on actual creation, which is correct. If a construction fails and computeIfAbsent is retried, the next call will get a new index — this is fine since failed instances are discarded.

LGTM. Well-designed fix with thorough cleanup and excellent test coverage.

---

Automated review by github-manager-bot

[qianye1001]

CI follow-up

The original JDK 8 integration failure was unrelated to the MQClientManager production change, but it exposed two real timing assumptions in existing integration tests. Commit 52184f84f1 stabilizes them without weakening their assertions:

• QueryMsgByKeyIT now waits for the asynchronously built message index and retries only the expected MQClientException that reports no indexed message yet.
• BatchAckIT now waits until all sent messages are visible in the consume queues before the first orderly POP. This prevents a partially dispatched queue from being order-locked after returning only part of the batch. Its POP assertion also allows multiple 3-second long-poll attempts on a busy runner.

A later macOS/JDK 8 Build run exposed a separate flaky ServiceThreadTest: it classified any wakeup taking 18-20 ms as lost, so one OS scheduling delay failed the test. Commit 1385ac7975 replaces that performance threshold with a barrier-based protocol. One waiter and four wakers coordinate every iteration; all 1,000 notified waits must complete, while an actually lost wakeup breaks the 20-second coordination bound.

Validation on the final head 1385ac7975:

• Build with Maven / macOS JDK 8: passed in 36m14s
• Build with Maven / Ubuntu JDK 8: passed in 40m35s
• Build with Maven / Windows JDK 8: passed in 44m27s
• Integration Tests / Ubuntu JDK 8: passed in 25m33s
• Coverage, CodeQL, Bazel, License, and Misspell: passed

Local Oracle JDK 8 validation also passed MQClientManagerTest (7/7), QueryMsgByKeyIT (4/4), the formerly failing orderly BatchAckIT path (1/1), and ServiceThreadTest twice (7/7 each run).

[RockteMQ-AI]

Summary

This PR modifies 6 file(s) with 961 lines of changes.

Review Notes

• Files changed: client/src/main/java/org/apache/rocketmq/client/impl/MQClientManager.java,client/src/main/java/org/apache/rocketmq/client/impl/factory/MQClientInstance.java,client/src/test/java/org/apache/rocketmq/client/impl/MQClientManagerTest.java,common/src/test/java/org/apache/rocketmq/common/ServiceThreadTest.java,test/src/test/java/org/apache/rocketmq/test/client/consumer/pop/BatchAckIT.java ... and 1 more
• Diff size: 961 lines
• CLA status: unknown
• ✅ Tests included

Observations

• Potential null safety: 3 chained .get() calls without null checks

Verdict

Code changes look reasonable. No critical issues detected in the structural review.

---

Automated review by github-manager-bot