apache / apache/pulsar-client-go

Reader.Next(ctx) redelivers a same message after reconnecting to broker

Open
#605 0 comments 0 reactions 0 assignees View on GitHub
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.