apache / apache/rocketmq

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

Open
#10,935 5 comments 0 reactions 0 assignees View on GitHub
Dominant language
Java
Stars
22.6k
Forks
12k
Avg merge
3d 1h
Merged PRs (30d)
27

Description

## Before Creating the Bug Report

- [x] I found a bug, not just a question for GitHub Discussions.
- [x] I searched GitHub Issues and Discussions and believe this is not a duplicate.
- [x] I confirmed that this bug belongs to the `apache/rocketmq` repository.

## 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

With a 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.

Contributor guide

Open the contributing guide

Research direction

Start with ProduceAccumulator.start()/shutdown() and the DefaultMQProducer lifecycle described in the report, then inspect the shared accumulator behavior through MQClientManager. Run the focused lifecycle reproduction and the ProduceAccumulatorTest and DefaultMQProducerTest suites. Done means one producer shutdown leaves timeout guards active for another same-client producer, while the final shutdown stops them and existing tests remain green.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
backend, distributed-systems
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Active
Clarity
Clearly specified
Newbie friendliness
68/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.