[Bug] Messages lost if a multi-topics consumer does not receive all messages after seek
- Dominant language
- Java
- Stars
- 15.3k
- Forks
- 3.8k
- Avg merge
- 1d 14h
- Merged PRs (30d)
- 160
Description
### Search before asking
- [X] I searched in the [issues](https://github.com/apache/pulsar/issues) and found nothing similar.
### Version
master (207335a449f2bc9cdf6782c67f93f8c2fb267271)
### Minimal reproduce step
Add `Thread.sleep(100);` after https://github.com/apache/pulsar/blob/e2f94dc98dbecb4dc401ba837c54f497ca9d896f/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MultiTopicsConsumerTest.java#L344
Then, run testSeekCustomTopicMessageId
### What did you expect to see?
The test should pass
### What did you see instead?
The test is blocked because not all messages are received.
### Anything else?
The reason is the current `seek` behavior of a multi-topics consumer.
Since https://github.com/apache/pulsar/pull/19158, a `TopicMessageId` can be accepted by a `seek` call on a multi-topics consumer. In this case, the consumer will perform seek operation on a specific internal consumer and other internal consumers are not affected.
However, since all incoming messages will be cleared before a seek operation, there is a race that some of them are not received by consumer before the next seek operation.
For example, given a multi-topics consumer that subscribes `topic1`, `topic2`, `topic3` and each topic has 5 messages.
- topic1: 0,1,2
- topic2: 3,4,5
1. Consumer seeks to 1 on topic1. Then the internal consumer will receiving from 2.
2. Wait for a while, `incomingMessages` might become `[2]` because internal consumer on topic1 could pre-fetch the message `2`.
3. Consumer seeks to 4 on topic2.
After that, consumer could only receive `[5]` because message `2` was pre-fetched in step 2 and cleared in `topic2`.
From expectations of users, they have sought to `topic1: 1` and `topic2:4`, so after that, they should receive `[2, 5]` rather than `[5]`.
### Are you willing to submit a PR?
- [x] I'm willing to submit a PR!
Contributor guide
Research direction
Start with pulsar-broker/src/test/java/org/apache/pulsar/client/api/MultiTopicsConsumerTest.java at testSeekCustomTopicMessageId and reproduce the issue by adding the stated sleep before running the test. Then trace multi-topics consumer seek handling and how incomingMessages are cleared across internal consumers. Done means the test passes and messages already prefetched for other sought topics are not lost.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- distributed-systems
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 42/100