[Bug] Bucket delayed delivery recovery can block dispatcher threads
- Dominant language
- Java
- Stars
- 15.3k
- Forks
- 3.8k
- Avg merge
- 1d 14h
- Merged PRs (30d)
- 160
Description
### Search before reporting
- [x] I searched in the issues and found nothing similar.
### Read release policy
- [x] I understand that unsupported versions don't get bug fixes. I will
attempt to reproduce the issue on a supported version of Pulsar client and
Pulsar broker.
### User environment
- Broker version: current master
- Delayed delivery tracker: bucket delayed delivery tracker
### Issue Description
`BucketDelayedDeliveryTracker` is lazily created from the broker dispatch path
when the first delayed message needs a delayed delivery tracker.
During creation, the bucket tracker recovers delayed bucket snapshots and waits
for recovery synchronously:
```text
PersistentDispatcherMultipleConsumers.trackDelayedDelivery()
-> delayedDeliveryTrackerFactory.newTracker(this)
-> new BucketDelayedDeliveryTracker(...)
-> recoverBucketSnapshot()
-> FutureUtil.waitForAll(...).get(...)
```
This can block a dispatcher thread while holding the dispatcher lock. If bucket
snapshot recovery is slow, dispatch for the current subscription is stalled, and
other tasks queued on the same `broker-topic-workers-*` thread can also be
delayed.
### Reproducing the issue
1. Enable bucket delayed delivery tracker.
2. Use a subscription that has delayed bucket snapshots to recover.
3. Make bucket snapshot metadata or storage operations slow.
4. Trigger dispatch of a delayed message after the tracker has not yet been
created, such as after broker restart or topic unload/load.
5. Observe that the dispatch path waits for bucket snapshot recovery before it
can continue.
### Additional information
This issue is about synchronous bucket snapshot recovery during lazy delayed
delivery tracker creation from the dispatch path.
### Are you willing to submit a PR?
- [x] I'm willing to submit a PR!
Contributor guide
Research direction
Start with PersistentDispatcherMultipleConsumers.trackDelayedDelivery(), then follow delayedDeliveryTrackerFactory.newTracker(this) into BucketDelayedDeliveryTracker and recoverBucketSnapshot(). Reproduce with delayed bucket snapshots and deliberately slow metadata or storage operations. Done means lazy tracker creation no longer synchronously blocks dispatcher threads or holds the dispatcher lock during bucket snapshot recovery.
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
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 48/100