[ISSUE #10935] Fix shared produce accumulator lifecycle

[ai-yang]

Which Issue(s) This PR Fixes

• Fixes [Bug] Shutting down one producer stops timeout flushing for producers with the same client ID #10935

Brief Description

MQClientManager can return the same ProduceAccumulator to multiple producers that share a RocketMQ client ID. Previously, every DefaultMQProducer.shutdown() stopped that shared accumulator, even if another producer using it was still running. Under low traffic, its small synchronous auto-batched sends could then wait indefinitely, while asynchronous messages remained queued without a callback because timeout flushing had stopped.

This change:

• reference-counts started producer owners inside ProduceAccumulator;
• starts the sync and async guard services only for the first owner;
• stops the guard services only after the last owner shuts down;
• tracks accumulator ownership in each DefaultMQProducer with an AtomicBoolean, making accumulator release idempotent across repeated shutdown calls;
• adds regression coverage for both shared lifetime and repeated shutdown.

There is no public API or wire-format change.

How Did You Test This Change?

The new lifecycle regression was first applied to the unmodified develop baseline. It wires the same accumulator into two real DefaultMQProducer instances, starts both through the public lifecycle methods, shuts down one producer, queues a small asynchronous message, and waits for its callback. Two baseline runs failed deterministically after the callback deadline:

Tests run: 1, Failures: 1
Expecting value to be true but was false
BUILD FAILURE

With this change, the complete ProduceAccumulatorTest suite passes:

Tests run: 5, Failures: 0, Errors: 0, Skipped: 0
BUILD SUCCESS

I also ran all existing DefaultMQProducerTest cases together with the accumulator tests:

Tests run: 46, Failures: 0, Errors: 0, Skipped: 0
BUILD SUCCESS

Command:

mvn -o -Dmaven.repo.local=/tmp/rocketmq-bug-m2 \
  -pl client -am -DskipITs \
  -Dcheckstyle.skip -Dspotbugs.skip -Drat.skip \
  -Dsurefire.failIfNoSpecifiedTests=false \
  -Dtest=ProduceAccumulatorTest,DefaultMQProducerTest test

git diff --check also passes.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 87.50000% with 1 line in your changes missing coverage. Please review.
✅ Project coverage is 48.50%. Comparing base (293f588) to head (4ce9e01).

Files with missing lines	Patch %	Lines
...he/rocketmq/client/producer/DefaultMQProducer.java	66.66%	0 Missing and 1 partial ⚠️

Additional details and impacted files

@@              Coverage Diff              @@
##             develop   #10937      +/-   ##
=============================================
- Coverage      48.62%   48.50%   -0.13%     
+ Complexity     13692    13650      -42     
=============================================
  Files           1381     1381              
  Lines         101464   101468       +4     
  Branches       13187    13187              
=============================================
- Hits           49337    49215     -122     
- Misses         46142    46237      +95     
- Partials        5985     6016      +31     

☔ 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

Fixes a real bug where MQClientManager returns the same ProduceAccumulator to multiple producers sharing a client ID. Previously, shutting down one producer would stop the shared accumulator, breaking the other producer's batch sends. This PR adds reference counting in ProduceAccumulator and AtomicBoolean idempotency in DefaultMQProducer.

Findings

• [Info] client/src/main/java/org/apache/rocketmq/client/producer/ProduceAccumulator.java:160–170 — The dual protection (AtomicBoolean in DefaultMQProducer + synchronized + producerCount in ProduceAccumulator) is correct and serves distinct purposes: the AtomicBoolean prevents a single producer from calling start/shutdown multiple times, while producerCount handles multiple producers sharing the same accumulator. Well-designed.
• [Info] client/src/test/java/org/apache/rocketmq/client/producer/ProduceAccumulatorTest.java:115–168 — The regression test testSharedAccumulatorRemainsRunningUntilLastProducerShutdown correctly reproduces the bug on the unmodified baseline (deterministic failure) and passes with the fix. Good use of CountDownLatch with timeout for async verification.

Verdict

Clean bug fix with proper synchronization and comprehensive regression tests. The reference counting approach is the right solution for shared accumulator lifecycle management.

---

Automated review by github-manager-bot