[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 while another producer was running. Under low traffic, small synchronous auto-batched sends could then wait indefinitely, while asynchronous messages remained queued without timeout flushing.

Keeping the guard threads alive is necessary but not sufficient: a pending batch also retains the producer that created it. A second producer must not join that batch after the first producer shuts down.

This change:

• reference-counts started producer owners; starts the guards for the first owner and stops them after the last owner;
• tracks each producer's ownership with an AtomicBoolean, making repeated shutdown release idempotent;
• includes the concrete producer's identity in both synchronous and asynchronous aggregation keys, including explicit-queue sends; a replacement instance with the same group is still a different owner;
• releases held bytes exactly once after send completion or a synchronous send-initiation failure, including when a user failure callback throws;
• removes the exact completed batch from the maps so producer-specific keys do not retain completed batches or remove a newer replacement.

There is no public API or wire-format change.

Applicability and tradeoffs

Automatic batching is disabled by default (autoBatch=false). The shared-lifecycle failure requires producers sharing an accumulator and one shutting down while another remains active; it is not a claim that ordinary non-batched sends are affected.

Producer identity intentionally prevents cross-producer coalescing, because producers may have different executors, hooks, credentials, and lifetimes. Batching within one producer is preserved. The aggregation/throughput tradeoff has not been benchmarked. This does not promise graceful draining of the producer that is shutting down, nor redesign user-callback exception/notification semantics.

How Did You Test This Change?

Latest local verification (2026-09-13), head f3901017f89509118835cd90b40109b3b51aa9c3, JDK 8 / Maven 3.8.1:

mvn -pl client -am -DskipITs \
  -Dtest=ProduceAccumulatorTest,DefaultMQProducerTest,ProduceAccumulatorLifecycleTest \
  -Dsurefire.failIfNoSpecifiedTests=false clean test

The local run used offline mode and settings pointing to the existing dependency cache; no project build configuration was changed and no Checkstyle/SpotBugs skip flags were used.

• ProduceAccumulatorTest: 5 passed, including real guard lifetime and repeated shutdown.
• DefaultMQProducerTest: 41 passed.
• ProduceAccumulatorLifecycleTest: 7 passed: stopped/running producer isolation, explicit queue, same-group replacement, completed-map cleanup, synchronous initiation failure, inline callback followed by throw, and throwing failure callback.
• Total: 53 tests, 0 failures/errors/skips; all four reactor modules succeeded.
• Checkstyle and SpotBugs enabled; git diff --check passed.

The new ownership tests disable background guards and explicitly flush real batches, making batch ownership deterministic; the existing accumulator suite separately exercises guard lifecycle. Re-running these seven new cases against the previous PR's production classes (ce2e9b171) produced 6 failures / 7 tests; the updated production classes passed all seven. This comparison is against the previous PR head, not a fresh full-suite run of current upstream develop.

The original guard-lifetime regression also failed on unmodified develop at e348efa66 before the reference-count fix. The PR retains that base and adds a fast-forward follow-up commit; it was not rebased. CI for the new head must be checked independently of these local results.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 91.66667% with 3 lines in your changes missing coverage. Please review.
✅ Project coverage is 49.32%. Comparing base (e348efa) to head (f390101).
⚠️ Report is 24 commits behind head on develop.

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

Additional details and impacted files

@@              Coverage Diff              @@
##             develop   #10937      +/-   ##
=============================================
+ Coverage      48.59%   49.32%   +0.72%     
- Complexity     13680    14222     +542     
=============================================
  Files           1381     1390       +9     
  Lines         101475   103134    +1659     
  Branches       13190    13486     +296     
=============================================
+ Hits           49313    50872    +1559     
+ Misses         46163    46085      -78     
- Partials        5999     6177     +178     

☔ 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

[RockteMQ-AI]

Summary

Fixes a lifecycle bug where multiple DefaultMQProducer instances sharing the same ProduceAccumulator (via same client ID) would cause IllegalThreadStateException or premature thread shutdown on start()/shutdown().

Findings

• [Info] client/src/main/java/org/apache/rocketmq/client/producer/ProduceAccumulator.java — Reference counting with synchronized is the correct pattern for shared guard threads. start() only launches threads on first producer; shutdown() only stops them when the last producer shuts down.
• [Info] client/src/main/java/org/apache/rocketmq/client/producer/DefaultMQProducer.java — AtomicBoolean.compareAndSet() ensures idempotent start/shutdown per producer instance, preventing double-start of the shared accumulator.
• [Info] Tests cover shared lifecycle and idempotent behavior — good coverage.

LGTM.

---

Automated review by github-manager-bot

[RockteMQ-AI]

Summary

Fixes the lifecycle management of shared ProduceAccumulator instances. When multiple DefaultMQProducer instances share the same accumulator (by client ID), the accumulator's background threads were incorrectly started/stopped by each producer independently.

Changes reviewed:

• DefaultMQProducer: Adds AtomicBoolean produceAccumulatorStarted to ensure start()/shutdown() on the accumulator are called exactly once per producer lifecycle.
• ProduceAccumulator: Adds producerCount with synchronized start()/shutdown() — reference counting ensures threads only start on first producer and stop when the last producer shuts down.
• Tests verify: shared accumulator remains running until last producer shuts down, and double-shutdown is idempotent.

Assessment:

• ✅ Correctness: Fixes a real bug where shared accumulator threads could be prematurely stopped or double-started.
• ✅ Thread safety: synchronized on accumulator start/shutdown, AtomicBoolean for producer-side guard.
• ✅ Tests: Good coverage of shared lifecycle and double-shutdown scenarios.
• ✅ Backward compatible: Single producer behavior is unchanged.

LGTM — correct fix for a subtle shared-state lifecycle bug.

---

Automated review by github-manager-bot

[ai-yang]

Update after refreshing this PR against the latest develop:

• rebased onto e348efa66b08eb645ee123706ea6492fa9a3ad35 (RocketMQ 5.5.1);
• extended the lifecycle regression to call start() twice and shutdown() twice on the same producer, verifying that the shared accumulator is acquired and released exactly once;
• ProduceAccumulatorTest and DefaultMQProducerTest: 46/46 passed;
• Checkstyle, SpotBugs, and git diff --check passed in the affected reactor build.

The force-push has retriggered the full CI matrix and should also exercise the previously partial DefaultMQProducer.start() branch in patch coverage.

@guyinyou, since you have context in ProduceAccumulator, could you please take a human review when available, especially around the reference-counted lifecycle and repeated start/shutdown behavior? Thank you.

[RockteMQ-AI]

Summary

Rebased fix for shared ProduceAccumulator lifecycle — looks good.

The two-level protection is sound:

• Per-producer (AtomicBoolean.compareAndSet): prevents double start/stop from the same producer instance
• Shared accumulator (synchronized producerCount): prevents premature shutdown when multiple producers share the same accumulator through MQClientManager

Test coverage validates both the shared-lifecycle and idempotent-start/stop scenarios.

LGTM.

---

Automated review by github-manager-bot

[ai-yang]

All 10 CI checks are green on the refreshed single signed-off commit. @RongtongJin @drpmma, could you please take a human review when convenient? The key point is the two-level lifecycle protection: per-producer idempotence plus reference-counted start/stop of the accumulator shared through MQClientManager.

[ai-yang]

Pushed follow-up commit f390101 to complete the shared-accumulator lifecycle fix.

The guard reference count alone did not isolate pending batches: a running producer could join a batch retaining an already stopped producer. Aggregation keys now include the concrete producer identity (also for explicit queues and same-group replacement instances). Completion/failure releases held bytes once and removes only that completed batch, avoiding stale producer retention.

Validation: 53 targeted client tests passed (5 accumulator + 41 producer + 7 new lifecycle cases), with Checkstyle and SpotBugs enabled. The seven new cases produced six failures against the previous PR production classes and all passed with this follow-up. The description documents the deterministic test setup and the autoBatch=false default.

@guyinyou, could you review the new batch ownership and cleanup semantics when convenient? In particular, producer isolation deliberately trades cross-producer coalescing for lifecycle correctness; no throughput improvement or graceful draining of a stopped producer is claimed. CI on this new head is separate from the earlier green result.

[RockteMQ-AI]

Summary

Re-reviewed after new commits. The reference-counting approach for shared ProduceAccumulator instances is well-designed and fixes a real lifecycle bug where shared accumulators could be prematurely shut down.

Key points:

• AtomicBoolean produceAccumulatorStarted ensures idempotent start/shutdown at the producer level
• producerCount reference counting in ProduceAccumulator correctly prevents the internal executor from being stopped while other producers still depend on it
• Thread safety: synchronized on start()/shutdown() in ProduceAccumulator is appropriate given the reference count mutation
• Test coverage validates the shared accumulator scenario

LGTM — re-approved with the latest changes.

---

Automated review by github-manager-bot