[GitHub] [pulsar] MarvinCai commented on a diff in pull request #17443: [fix][client]Duplicate messages when use MultiTopicsConsumerImpl

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

MarvinCai commented on code in PR #17443:
URL: https://github.com/apache/pulsar/pull/17443#discussion_r967621298


##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java:
##########
@@ -789,20 +804,31 @@ public CompletableFuture<Void> seekAsync(MessageId messageId) {
                     new PulsarClientException("Illegal messageId, messageId can only be earliest/latest")
             );
         }
-        List<CompletableFuture<Void>> futures = new ArrayList<>(consumers.size());
-        consumers.values().forEach(consumerImpl -> futures.add(consumerImpl.seekAsync(targetMessageId)));
-
-        unAckedMessageTracker.clear();
-        clearIncomingMessages();
-
-        return FutureUtil.waitForAll(futures);
+        return internalSeekAsync(consumer -> consumer.seekAsync(messageId));
     }
 
     @Override
     public CompletableFuture<Void> seekAsync(long timestamp) {
-        List<CompletableFuture<Void>> futures = new ArrayList<>(consumers.size());
-        consumers.values().forEach(consumer -> futures.add(consumer.seekAsync(timestamp)));
-        return FutureUtil.waitForAll(futures);
+        return internalSeekAsync(consumer -> consumer.seekAsync(timestamp));
+    }
+
+    private CompletableFuture<Void> internalSeekAsync(Function<Consumer, CompletableFuture<Void>> childSeekFunction) {
+        pause();
+        CompletableFuture<Void> res = FutureUtil.waitForAll(receiveMessageFutures).thenCompose(__ -> {

Review Comment:
   do we need to wait for completion of these ongoing receives as the messages will be discarded anyway, can we just cancel them?



-- 
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