[GitHub] [pulsar] lhotari commented on a diff in pull request #20032: [fix][client] Fix client redeliver epoch bigger than broker consumer epoch

["lhotari (via GitHub)" <gi...@apache.org>]

lhotari commented on code in PR #20032:
URL: https://github.com/apache/pulsar/pull/20032#discussion_r1189700517


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -1812,20 +1812,31 @@ protected void handleRedeliverUnacknowledged(CommandRedeliverUnacknowledgedMessa
                     remoteAddress, redeliver.getConsumerId(),
                     redeliver.hasConsumerEpoch() ? redeliver.getConsumerEpoch() : null);
         }
-
-        CompletableFuture<Consumer> consumerFuture = consumers.get(redeliver.getConsumerId());
-
-        if (consumerFuture != null && consumerFuture.isDone() && !consumerFuture.isCompletedExceptionally()) {
-            Consumer consumer = consumerFuture.getNow(null);
-            if (redeliver.getMessageIdsCount() > 0 && Subscription.isIndividualAckMode(consumer.subType())) {
-                consumer.redeliverUnacknowledgedMessages(redeliver.getMessageIdsList());
-            } else {
-                if (redeliver.hasConsumerEpoch()) {
-                    consumer.redeliverUnacknowledgedMessages(redeliver.getConsumerEpoch());
+        boolean hasConsumerEpoch = redeliver.hasConsumerEpoch();
+        List<MessageIdData> messageIdsList = redeliver.getMessageIdsList();
+        int messageIdsCount = redeliver.getMessageIdsCount();
+        long consumerId = redeliver.getConsumerId();
+        long consumerEpoch = redeliver.getConsumerEpoch();
+        CompletableFuture<Consumer> consumerFuture = consumers.get(consumerId);
+        if (consumerFuture != null) {
+            consumerFuture.thenAccept((consumer) -> {
+                if (messageIdsCount > 0 && Subscription.isIndividualAckMode(consumer.subType())) {
+                    consumer.redeliverUnacknowledgedMessages(messageIdsList);
                 } else {
-                    consumer.redeliverUnacknowledgedMessages(DEFAULT_CONSUMER_EPOCH);
+                    if (hasConsumerEpoch) {
+                        consumer.redeliverUnacknowledgedMessages(consumerEpoch);
+                    } else {
+                        consumer.redeliverUnacknowledgedMessages(DEFAULT_CONSUMER_EPOCH);
+                    }
                 }
-            }
+            }).exceptionally(e -> {
+                // if consumerFuture completed exceptionally, don't need to process this redeliver command
+                // because, consumer will reconnect
+                log.warn("[{}] ignore this redeliverUnacknowledged request from consumer {}, consumerEpoch {}",
+                        remoteAddress, redeliver.getConsumerId(),
+                        redeliver.hasConsumerEpoch() ? redeliver.getConsumerEpoch() : null, e);
+                return null;
+            });

Review Comment:
   Isn't this refactoring completely irrelevant in this PR? If the intention is to add a log statement, just add that without refactoring the existing logic. 



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