iloveeyjafjalla opened a new issue, #11441:
URL: https://github.com/apache/rocketmq/issues/11441
### Describe the Bug
When the smallest broker of a group changes and this broker becomes the
master, `BrokerController#onMinBrokerChange` ends with
`pullRequestHoldService.notifyMasterOnline()`, whose job is to immediately
re-execute every pull request that is currently suspended, so that clients do
not have to wait for their own suspend timeout after a master comes online.
That sweep assumes a bucket it has just seen as non-empty still holds
requests when it drains it:
```java
for (ManyPullRequest mpr : this.pullRequestTable.values()) {
if (mpr == null || mpr.isEmpty()) {
continue;
}
for (PullRequest request : mpr.cloneListAndClear()) {
```
`ManyPullRequest#cloneListAndClear()` returns `null` (not an empty list)
whenever the bucket is empty at drain time:
```java
public synchronized List<PullRequest> cloneListAndClear() {
if (!this.pullRequestList.isEmpty()) {
...
return result;
}
return null;
}
```
The buckets are drained by the message-arrival path on the store's dispatch
thread (`NotifyMessageArrivingListener` ->
`PullRequestHoldService#notifyMessageArriving`, which drains the same bucket
and replays the requests that did not match). An arrival that wins the race
between the `isEmpty()` check above and the drain makes the loop iterable
`null`, so the enhanced `for` throws a `NullPointerException`.
### Impact
1. The `NullPointerException` escapes `notifyMasterOnline()` into
`onMinBrokerChange`. The sweep is over, so every bucket that had not yet been
visited keeps its requests suspended: those clients sit out their full suspend
timeout instead of being woken immediately - exactly the stall this method
exists to prevent, and in exactly the failover window it is meant for.
2. The exception is thrown on the thread that updates the min-broker state
(including the `UPDATE_MIN_BROKER` admin path), so the caller sees a failure
for an operation that has otherwise already been applied.
The same class shows what the intended contract is: `notifyMessageArriving`
drains a bucket the same way and guards the result with `if (requestList !=
null)` before iterating, so the missing guard in `notifyMasterOnline` is an
oversight rather than a documented behaviour.
### How to Reproduce
- Suspend a pull request on some `topic@queueId` (long polling enabled, i.e.
the client sent the suspend flag and the pull returned `PULL_NOT_FOUND` longer
than the max. offset).
- Produce a message for that queue from another thread, so the dispatch
thread drains the same bucket through `notifyMessageArriving` in the same
instant as `notifyMasterOnline()`.
- The stricter, deterministic form of the race: the bucket reports itself as
non-empty and is drained before `notifyMasterOnline()` drains it, which is what
makes the sweep iterable `null`.
- Result with the current code: `java.lang.NullPointerException at
org.apache.rocketmq.broker.longpolling.PullRequestHoldService.notifyMasterOnline(PullRequestHoldService.java:191)`,
and the remaining buckets are not woken.
### Expected Behavior
A bucket that lost that race is skipped; the sweep continues and still wakes
the requests held for the other queues.
### Proposed Fix
Read the drained list into a local variable and skip the bucket when it
answers `null`, instead of iterating the call result directly:
```java
List<PullRequest> requestList = mpr.cloneListAndClear();
if (requestList == null) {
continue;
}
for (PullRequest request : requestList) {
```
### Environment
- 5.5.1-SNAPSHOT (`develop`, 0dd48c39db), all platforms
--
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]
