ai-yang opened a new issue, #10935: URL: https://github.com/apache/rocketmq/issues/10935
### Before Creating the Bug Report - [x] I found a bug, not just asking a question, which should be created in [GitHub Discussions](https://github.com/apache/rocketmq/discussions). - [x] I have searched the [GitHub Issues](https://github.com/apache/rocketmq/issues) and [GitHub Discussions](https://github.com/apache/rocketmq/discussions) of this repository and believe that this is not a duplicate. - [x] I have confirmed that this bug belongs to the current repository, not other repositories of RocketMQ. ### Runtime platform environment Reproduced on Ubuntu 22.04.4 LTS, Linux 5.15.0-186, x86_64. The defect is in the Java client lifecycle and is not expected to be OS-specific. ### RocketMQ version - Branch: `develop` - Version: `5.5.0` / current develop sources - Commit: `293f5885719fc4aa3619446a1900f58ccfcfdd29` ### JDK Version - OpenJDK 8u492 ### Describe the Bug Proposed severity: **Major availability impact**. Multiple `DefaultMQProducer` instances with the same RocketMQ client ID share one `ProduceAccumulator` through `MQClientManager`. The client ID is based on client IP, `instanceName`, and unit name, and does not include the producer group. However, each producer currently starts and shuts down that shared accumulator independently. The service threads ignore repeated starts, while the first producer shutdown stops both the synchronous and asynchronous batch guard threads even when another producer using the accumulator is still running. After that shutdown: - under low traffic, a small synchronous auto-batched message can wait indefinitely because it does not reach the size threshold and the timeout guard no longer wakes it; - a small asynchronous auto-batched message can remain queued without a callback until another send reaches the size threshold. The surviving producer remains in `RUNNING` state, so the failure is silent and gives the application no useful error to recover from. This is an availability bug, not a security report. ### Steps to Reproduce reachable NameServer and Broker, create two auto-batching producers in different producer groups but give them the same explicit instance name. Ensure `TopicTest` already exists or topic auto-creation is enabled: ```java DefaultMQProducer producerA = new DefaultMQProducer("group-a"); DefaultMQProducer producerB = new DefaultMQProducer("group-b"); producerA.setNamesrvAddr(namesrvAddr); producerB.setNamesrvAddr(namesrvAddr); producerA.setInstanceName("shared-client"); producerB.setInstanceName("shared-client"); producerA.setAutoBatch(true); producerB.setAutoBatch(true); producerA.batchMaxDelayMs(1000); producerB.batchMaxDelayMs(1000); producerA.start(); producerB.start(); producerA.shutdown(); // A small message is below the auto-batch size threshold. This call can // block indefinitely because the shared timeout guard was stopped by A. producerB.send(new Message("TopicTest", new byte[] {1})); ``` The same defect has a deterministic broker-free unit reproduction: 1. Create one `ProduceAccumulator` with a short `batchMaxDelayMs`. 2. Call `start()` twice to represent two producers sharing the same client ID. 3. Call `shutdown()` once to release only the first producer. 4. Add one small asynchronous message through `MockMQProducer`. 5. Wait longer than `batchMaxDelayMs` for either callback. On the unmodified baseline, the callback deadline expires every time. Two independent runs of the final public-lifecycle regression failed with method times of 4.817 and 4.829 seconds: ```text Expecting value to be true but was false Tests run: 1, Failures: 1 ``` ### What Did You Expect to See? Shutting down one producer should release only that producer's ownership of the shared accumulator. As long as another producer with the same client ID is still started, the accumulator guard threads should remain active and flush its messages after `batchMaxDelayMs`. ### What Did You See Instead? The first producer shutdown stops the shared guard threads immediately. Under low traffic, a remaining producer can then block indefinitely in a synchronous send, or retain an asynchronous message without invoking its callback. A later same-key send that reaches the size threshold can still trigger a size-based flush; the broken behavior is timeout flushing. ### Additional Context A minimal fix is to reference-count started producer owners in `ProduceAccumulator`: start its guards only on transition `0 -> 1` and stop them only on `1 -> 0`. Each `DefaultMQProducer` should retain and release at most once so repeated `shutdown()` calls cannot decrement another producer's ownership. With that fix, the new focused suite passes 5/5 tests, and the broader `ProduceAccumulatorTest` plus `DefaultMQProducerTest` run passes 46/46 tests. Temporary workarounds are to use a unique `instanceName` for each producer or to avoid shutting down any producer that shares the client ID until all such producers are ready to stop. I searched for reports involving `ProduceAccumulator`, its sync/async guard threads, shared accumulator lifecycle, same `instanceName`, and producer shutdown, and found no duplicate. Related issue [#8806](https://github.com/apache/rocketmq/issues/8806) and its merged fix [#8807](https://github.com/apache/rocketmq/pull/8807) address accidental sharing caused by early initialization. They intentionally retain sharing for producers with the same instance/unit identity and do not manage ownership of the shared accumulator's lifecycle; this report concerns that distinct lifecycle problem. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
