[GitHub] [pulsar] mattisonchao commented on a diff in pull request #17728: [fix][tableview] fixed ack failure in ReaderImpl due to null messageId

[GitBox <gi...@apache.org>]

mattisonchao commented on code in PR #17728:
URL: https://github.com/apache/pulsar/pull/17728#discussion_r979515227


##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderImpl.java:
##########
@@ -177,17 +178,20 @@ public Message<T> readNext(int timeout, TimeUnit unit) throws PulsarClientExcept
 
     @Override
     public CompletableFuture<Message<T>> readNextAsync() {
-        CompletableFuture<Message<T>> receiveFuture = consumer.receiveAsync();
-        receiveFuture.whenComplete((msg, t) -> {
-           if (msg != null) {
-               consumer.acknowledgeCumulativeAsync(msg).exceptionally(ex -> {
-                   log.warn("[{}][{}] acknowledge message {} cumulative fail.", getTopic(),
-                           getConsumer().getSubscription(), msg.getMessageId(), ex);
-                   return null;
-               });
-           }
+        CompletableFuture<Message<T>> originalFuture = consumer.receiveAsync();
+        CompletableFuture<Message<T>> result = originalFuture.thenApply(msg -> {
+            consumer.acknowledgeCumulativeAsync(msg)

Review Comment:
   > see ack failures due to null message id -- because we release the msg in TableViewImpl before reader.acknowledgeCumulativeAsync.
   
   This fix will still exist problem (logically) because `consumer.acknowledgeCumulativeAsync(msg)` is an async operation. It may run parallel with the user operation.
   Please let me know what you think or if I'm missing something. thanks a lot!



-- 
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: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org