RockteMQ-AI commented on issue #10935: URL: https://github.com/apache/rocketmq/issues/10935#issuecomment-5370289516
Hi @ai-yang, thanks for reporting this! I've analyzed this issue against the codebase and can confirm this appears to be a **real bug**. **Root Cause:** ProduceAccumulator is shared per clientId by MQClientManager, but ProduceAccumulator.start()/shutdown() and DefaultMQProducer's calls to them are not reference-counted. The first producer shutdown stops the shared guard threads even when another producer using the same accumulator is still running, so timeout-based flushing ceases. **Affected Files:** client/src/main/java/org/apache/rocketmq/client/producer/ProduceAccumulator.java, client/src/main/java/org/apache/rocketmq/client/producer/DefaultMQProducer.java, client/src/main/java/org/apache/rocketmq/client/impl/MQClientManager.java **Analysis:** MQClientManager.getOrCreateProduceAccumulator() (client/src/main/java/org/apache/rocketmq/client/impl/MQClientManager.java:68) returns the same ProduceAccumulator for producers with the same clientId (IP + instanceName + unitName, no producer group). DefaultMQProducer.initProduceAccumulator() (DefaultMQProducer.java:1475) obtains that shared instance. On start, DefaultMQProducer.start() (DefaultMQProducer.java:377-379) calls produceAccumulator.start(), and on shutdown DefaultMQProducer.shutdown() (DefaultMQProducer.java:414-416) calls produceAccumulator.shutdown(). ProduceAccumulator.start() (ProduceAccumulator.java:158-161) starts two ServiceThreads, and ProduceAccumulator.shutdown() (ProduceAccumulator.java:163-166) stops them. ServiceThread.start() ignores repeated starts, while ServiceThread.shutdown() unconditionally stops the thread and marks it not started. Therefore, after two producers share the accumulator and the first one shuts down, the guard threads are stopped even though the second producer is still RUNNING, causing small auto-batched messages to wait indefinitely for the timeout flush. I'll prepare a fix spec and work on a PR. The community is welcome to provide feedback on the approach before implementation. --- 🤖 *Automated issue analysis by github-manager* -- 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]
