rueian opened a new issue #605:
URL: https://github.com/apache/pulsar-client-go/issues/605
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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
