gogodjzhu opened a new issue, #11309:
URL: https://github.com/apache/rocketmq/issues/11309
### Version
Current `develop` branch.
### Description
`DefaultMQPushConsumer.shutdown()` can race with asynchronous rebalance work
performed by the shared `MQClientInstance`.
The asynchronous rebalance path is:
```text
RebalanceService
→ MQClientInstance.doRebalance()
→ MQConsumerInner.tryRebalance()
→ DefaultMQPushConsumerImpl.tryRebalance()
→ RebalanceImpl.doRebalance()
```
Before this fix, `shutdown()` and `doRebalance()` were serialized by the
consumer instance lock, but the actual asynchronous entry point,
`tryRebalance()`, was neither synchronized nor guarded by the consumer
lifecycle state.
As a result, an asynchronous rebalance can still be in progress while the
consumer shuts down, or can run after shutdown releases the consumer lock:
1. `shutdown()` destroys `RebalanceImpl`, marks existing `ProcessQueue`
instances as dropped, and clears `processQueueTable`;
2. a concurrent or late `tryRebalance()` adds a new `ProcessQueue` to the
table;
3. rebalance dispatches a `PullRequest` referencing that `ProcessQueue` to
`PullMessageService`;
4. the consumer lifecycle has already ended, so the newly created
`ProcessQueue` no longer has a normal rebalance-management or cleanup
path.
This leaves an orphan `ProcessQueue`: it can still be referenced by a queued
`PullRequest`, but is no longer managed by the consumer's current
`processQueueTable`.
### Impact
An orphan `ProcessQueue` can cause:
- **Unnecessary pulls and resource leakage**: a `PullRequest` referencing the
orphan queue can remain in the shared `PullMessageService` queue. Since the
queue was not marked as dropped, the pull path cannot discard it
immediately.
- **Missing lifecycle cleanup**: because the queue is not reliably managed by
`processQueueTable`, subsequent rebalance and normal shutdown cleanup
cannot
reliably find and mark it as dropped.
- **Inconsistent message-processing state**: stale pull and consume work can
become detached from the consumer lifecycle, potentially causing useless
consume scheduling, offset-handling anomalies, or duplicate-processing
risk.
- **Higher risk during frequent restart**: when a shared `MQClientInstance`
remains alive for other producers or consumers, `RebalanceService`
continues
running and rapid consumer restarts make this race more likely.
### Reproduction
1. Create a `DefaultMQPushConsumer` registered with a shared
`MQClientInstance`.
2. Let `RebalanceImpl.doRebalance()` enter the path that creates a
`ProcessQueue` and dispatches a `PullRequest`.
3. Call `DefaultMQPushConsumer.shutdown()` concurrently while the
asynchronous
rebalance is in progress.
4. After shutdown, inspect `PullMessageService`:
- the `PullRequest` references a `ProcessQueue` that is absent from the
consumer's current `processQueueTable`; and
- the referenced `ProcessQueue` is not marked as dropped.
### Proposed Fix
Synchronize `DefaultMQPushConsumerImpl.tryRebalance()` with the same consumer
instance lock used by `shutdown()`, and check that the consumer is in the
`RUNNING` state before performing rebalance.
This ensures that:
- shutdown waits for a rebalance already in progress;
- rebalance arriving after shutdown is skipped because the consumer is no
longer `RUNNING`; and
- no new `ProcessQueue` or `PullRequest` is created after
`RebalanceImpl.destroy()`.
--
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]
