codelipenghui commented on code in PR #19031:
URL: https://github.com/apache/pulsar/pull/19031#discussion_r1059254154
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java:
##########
@@ -204,6 +208,10 @@
private final AtomicReference<ClientCnx>
clientCnxUsedForConsumerRegistration = new AtomicReference<>();
private final List<Throwable> previousExceptions = new
CopyOnWriteArrayList<Throwable>();
+ // Key is the ledger id and the entry id, entry is the acker that
represents which single messages are acknowledged
+ private final ConcurrentNavigableMap<Pair<Long, Long>, BatchMessageAcker>
batchMessageToAcker =
Review Comment:
> Regarding the message redelivering or reconnection, if the batch message
id was not created by deserialization, the acker inside the BatchMessageIdImpl
would not be modified. If we only clean up the batchMessageToAcker for
deserialized batch message id, the behavior would be different.
If the user decides to nack the message, they should not continue to ack it.
After the message redelivery, the newly received message with a new Acker. But
this PR introduced a shared state. It looks like the newly received message
with the different acker can also associate with the old ack state if we don't
clean up the shared state.
--
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]
