apache / apache/pulsar-client-go
Reader.Next(ctx) redelivers a same message after reconnecting to broker
- Dominant language
- Go
- Stars
- 745
- Forks
- 389
- Avg merge
- 3d 20h
- Merged PRs (30d)
- 3
Description
Hi, I found that the `Reader.Next(ctx)` redelivered a same message to me after reconnecting to broker and I thought it was not correct.
Consider the following reader setup:
```golang
reader, _ := c.CreateReader(pulsar.ReaderOptions{
Topic: "persistent://public/default/my_topic",
Name: "reader-test",
StartMessageID: pulsar.LatestMessageID(),
StartMessageIDInclusive: true,
})
for reader.HasNext() {
msg, _ := reader.Next(context.Background())
// msg is duplicated after reconnection
}
```
In my case, after receiving the `BaseCommand_CLOSE_CONSUMER` command from the broker, the `partitionConsumer` reconnected to broker and also recreated the non-durable subscription with `cmdSubscribe.StartMessageId = pc.lastDequeuedMsg`
https://github.com/apache/pulsar-client-go/blob/b6841513379ea9ca503d1e350c5f93198fc2b03f/pulsar/consumer_partition.go#L949-L953
And then I received a same message from the `reader.Next(context.Background())`.
I guess it was the `StartMessageIDInclusive: true` made it redeliver a same message again after reconnection. Maybe the `StartMessageIDInclusive` should always be changed to false in the case of Reader reconnection?
Another problem I noticed was that the accesses to `pc.lastDequeuedMsg` from the `reader` and `partitionConsumer` are not thread safe:
https://github.com/apache/pulsar-client-go/blob/b6841513379ea9ca503d1e350c5f93198fc2b03f/pulsar/reader_impl.go#L132
https://github.com/apache/pulsar-client-go/blob/b6841513379ea9ca503d1e350c5f93198fc2b03f/pulsar/consumer_partition.go#L1028-L1031
Contributor guide
Research direction
Start with the reconnection logic in pulsar/consumer_partition.go around lines 949-953 and the reader access in pulsar/reader_impl.go around line 132. Trace how cmdSubscribe.StartMessageId and StartMessageIDInclusive interact with pc.lastDequeuedMsg, then inspect the accesses around consumer_partition.go lines 1028-1031. Done means reconnecting a Reader no longer redelivers the same message and the shared state is accessed safely.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- go
- Domain
- backend-api-design, distributed-systems
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 35/100