codelipenghui commented on code in PR #19031:
URL: https://github.com/apache/pulsar/pull/19031#discussion_r1059392558
##########
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:
Users should not continue to process/ack the messages before the seek
operation. It's not the same case as I provided which is normal usage.
And we should ensure after the seek operation. We should not return the
cached messages before the seek operation to users. If users try to cache the
messages or just inflight messages, users need to guarantee they will not ack
the inflight or cached messages before the seek operation.
> BTW, if users left one message ID (messageIdList.get(4)) to acknowledge
after seek and they didn't acknowledge other message IDs, what will they
expect? Did they expect messages could be received again?
Yes, they should receive all the messages after the new seek position no
matter what happened before the seek operation.
--
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]
