Race condition with Reader.seek()
- Dominant language
- Java
- Stars
- 15.3k
- Forks
- 3.8k
- Avg merge
- 1d 22h
- Merged PRs (30d)
- 142
Description
**Describe the bug**
Consuming a message with the Reader interface immediately after seeking can consume the wrong message. This only seems to happen in a pretty specific situation (details below). Adding a short pause fixes it, so I wonder if it _could_ happen in other situations.
**To Reproduce**
IntelliJ / Maven project is attached: [PulsarSeek.zip](https://github.com/apache/pulsar/files/6523034/PulsarSeek.zip)
Adjust the `SERVICE_URL` as needed. I am using Pulsar 2.7.1 in Docker.
```java
public class SeekTest
{
private static final String SERVICE_URL = "pulsar://localhost:6650";
private static final String TEST_TOPIC = "persistent://public/default/seek_test";
public static void main(String[] args) throws Exception
{
PulsarClient client = PulsarClient.builder().serviceUrl(SERVICE_URL).build();
Producer producer = client.newProducer()
.topic(TEST_TOPIC)
.producerName("producer")
.create();
// Produce 100 messages and record the message IDs
System.out.print("Producing 100 messages... ");
ArrayList messageIds = new ArrayList<>();
for (int i = 0; i < 100; i++)
{
// Message is the loop counter
byte[] bytes = Integer.toString(i).getBytes(StandardCharsets.UTF_8);
MessageId id = producer.send(bytes);
messageIds.add(id);
}
System.out.println("Done");
Reader reader = client.newReader()
.topic(TEST_TOPIC)
.startMessageId(MessageId.latest)
.startMessageIdInclusive()
.create();
System.out.println("Seeking to message 50...");
reader.seek(messageIds.get(50));
reader.hasMessageAvailable();
Message message = reader.readNext();
System.out.println("Got MessageId: " + message.getMessageId());
System.out.println("Got message data: " + new String(message.getData(), StandardCharsets.UTF_8));
reader.closeAsync();
producer.closeAsync();
client.closeAsync();
}
}
```
**Expected behavior**
It should print "50", but it actually prints "99".
Output:
```
Producing 100 messages... Done
Seeking to message 50...
Got MessageId: 21:99:-1:0
Got message data: 99
```
**Additional context**
There are a few things that seem necessary to cause it to fail:
* Create Reader with `startMessageIdInclusive()` and `startMessageId(MessageId.latest)`
* Call `hasMessageAvailable()` after `seek()`
If you add a short pause between `seek()` and `hasMessageAvailable()`, it produces the correct result ("50"). This part is a bit worrying since it seems to be some type of race condition. Could it cause other problems that are harder to reproduce?
This can also be reproduced by seeking to a timestamp instead of a MessageId.
In my application, I avoid this issue by creating the Reader with an initial position (`startMessageId()` / `startMessageFromRollbackDuration()`) instead of using `seek()` and this works fine.
Contributor guide
Research direction
Start with the attached PulsarSeek.zip reproduction and the Reader.seek() and hasMessageAvailable() entry points, using Pulsar 2.7.1 in Docker as described. Verify the race by seeking to message 50 immediately before checking availability, then trace the reader state transition. Done means the reproduction consistently returns message 50 without requiring a pause and includes regression coverage.
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
- 45/100