Handle receiveAsync() failures in MultiTopicsConsumer (#11843)

apache · Aug 31, 2021 · 91bc3cc · 91bc3cc
1 parent 6486a55
commit 91bc3cc
Showing 1 changed file with 4 additions and 0 deletions.
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
@@ -269,6 +269,10 @@ private void receiveMessageFromConsumer(ConsumerImpl<T> consumer) {
                 // recursion and stack overflow
                 internalPinnedExecutor.execute(() -> receiveMessageFromConsumer(consumer));
             }
+        }).exceptionally(ex -> {
+            log.error("Receive operation failed on consumer {} - Retrying later", consumer, ex);
+            internalPinnedExecutor.schedule(() -> receiveMessageFromConsumer(consumer), 10, TimeUnit.SECONDS);
+            return null;
         });
     }