[fix][client] Fix duplicate messages caused by seek

Signed-off-by: Zixuan Liu <nodeces@gmail.com>

Author: nodece

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

@@ -2778,6 +2778,10 @@ public Map<MessageIdImpl, List<MessageImpl<T>>> getPossibleSendToDeadLetterTopic
         return possibleSendToDeadLetterTopicMessages;
     }
 
+    protected boolean isDuringSeek() {
+        return duringSeek.get();
+    }
+
     private static final Logger log = LoggerFactory.getLogger(ConsumerImpl.class);
 
 }

pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java

@@ -252,6 +252,10 @@ private void receiveMessageFromConsumer(ConsumerImpl<T> consumer, boolean batchR
             messagesFuture = consumer.receiveAsync().thenApply(Collections::singletonList);
         }
         messagesFuture.thenAcceptAsync(messages -> {
+            if (consumer.isDuringSeek()) {
+                receiveMessageFromConsumer(consumer, batchReceive);
+                return;
+            }
             if (log.isDebugEnabled()) {
                 log.debug("[{}] [{}] Receive message from sub consumer:{}",
                     topic, subscription, consumer.getTopic());