[Bug] lastMessageIdInBroker is earlier than lastDequeuedMessageId for topic policies system topic
- 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](https://github.com/apache/pulsar/issues) and found nothing similar.
### Read release policy
- [x] I understand that [unsupported versions](https://pulsar.apache.org/contribute/release-policy/#supported-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
4.0.5.3
### Issue Description
When I checked `SystemTopicBasedTopicPoliciesService#readerCaches`, I found a namespace's reader on `__change_events` topic has a strange state.
`incomingMessages` is empty, which means all messages have been received. However, `lastMessageIdInBroker` is `3501:2`, while `lastDequeuedMessageId` is `3721:4`. It's strange for this case, because the topic policies service uses a `hasMoreEventsAsync` and `readNextAsync` loop:
https://github.com/apache/pulsar/blob/a378bdcbe87efdc3b422f462192b04bb510179c3/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java#L460-L469
If the loop exits, the last call should be `hasMoreEventsAsync`, which must have retrieved the latest last message id as `lastMessageIdInBroker`
Internal stats:
```
}, {
"ledgerId" : 3501,
"entries" : 3,
"size" : 805,
"offloaded" : false,
"underReplicated" : false
}, {
"ledgerId" : 3620,
"entries" : 4,
"size" : 1078,
"offloaded" : false,
"underReplicated" : false
}, {
"ledgerId" : 3721,
"entries" : 5,
"size" : 1364,
"offloaded" : false,
"underReplicated" : false
}, {
```
### Error messages
```text
```
### Reproducing the issue
n/a
### Additional information
_No response_
### Are you willing to submit a PR?
- [ ] I'm willing to submit a PR!
Contributor guide
Research direction
Start in pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java around the hasMoreEventsAsync/readNextAsync loop at lines 460-469, then inspect readerCaches and the supplied internal topic statistics. Determine why lastMessageIdInBroker can lag behind lastDequeuedMessageId when incomingMessages is empty; done means the inconsistent state is explained and corrected, with behavior validated against the reported values.
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
- Stale
- Clarity
- Needs clarification
- Newbie friendliness
- 35/100