[fix][client] Handle synchronous RetryUtil supplier failures

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RetryUtilTest.java

@@ -24,6 +24,7 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import lombok.Cleanup;
 import org.apache.pulsar.client.util.RetryUtil;
@@ -60,6 +61,28 @@ public void testFailAndRetry() throws Exception {
         assertEquals(atomicInteger.get(), 5);
     }
 
+    @Test
+    public void testSupplierThrowsAndRetries() throws Exception {
+        @Cleanup("shutdownNow")
+        ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
+        CompletableFuture<Boolean> callback = new CompletableFuture<>();
+        AtomicInteger atomicInteger = new AtomicInteger(0);
+        Backoff backoff = Backoff.builder()
+                .initialDelay(Duration.ofMillis(1))
+                .maxBackoff(Duration.ofMillis(1))
+                .mandatoryStop(Duration.ofSeconds(5))
+                .jitterPercent(0)
+                .build();
+        RetryUtil.retryAsynchronously(() -> {
+            if (atomicInteger.incrementAndGet() < 3) {
+                throw new RuntimeException("fail");
+            }
+            return CompletableFuture.completedFuture(true);
+        }, backoff, executor, callback);
+        assertTrue(callback.get(5, TimeUnit.SECONDS));
+        assertEquals(atomicInteger.get(), 3);
+    }
+
     @Test
     public void testFail() throws Exception {
         @Cleanup("shutdownNow")

pulsar-client/src/main/java/org/apache/pulsar/client/util/RetryUtil.java

@@ -44,24 +44,34 @@ public static <T> void retryAsynchronously(Supplier<CompletableFuture<T>> suppli
     private static <T> void executeWithRetry(Supplier<CompletableFuture<T>> supplier, Backoff backoff,
                                              ScheduledExecutorService scheduledExecutorService,
                                              CompletableFuture<T> callback) {
-        supplier.get().whenComplete((result, e) -> {
-            if (e != null) {
-                long next = backoff.next().toMillis();
-                boolean isMandatoryStop = backoff.isMandatoryStopMade();
-                if (isMandatoryStop) {
-                    callback.completeExceptionally(e);
-                } else {
-                    log.warn().exceptionMessage(e)
-                            .attr("nextMs", next)
-                            .log("Execution with retry failed, will retry");
-                    scheduledExecutorService.schedule(() ->
-                                    executeWithRetry(supplier, backoff, scheduledExecutorService, callback),
-                            next, TimeUnit.MILLISECONDS);
+        try {
+            supplier.get().whenComplete((result, e) -> {
+                if (e != null) {
+                    retryOrCompleteExceptionally(supplier, backoff, scheduledExecutorService, callback, e);
+                    return;
                 }
-                return;
-            }
-            callback.complete(result);
-        });
+                callback.complete(result);
+            });
+        } catch (Throwable e) {
+            retryOrCompleteExceptionally(supplier, backoff, scheduledExecutorService, callback, e);
+        }
+    }
+
+    private static <T> void retryOrCompleteExceptionally(Supplier<CompletableFuture<T>> supplier, Backoff backoff,
+                                                         ScheduledExecutorService scheduledExecutorService,
+                                                         CompletableFuture<T> callback, Throwable e) {
+        long next = backoff.next().toMillis();
+        boolean isMandatoryStop = backoff.isMandatoryStopMade();
+        if (isMandatoryStop) {
+            callback.completeExceptionally(e);
+        } else {
+            log.warn().exceptionMessage(e)
+                    .attr("nextMs", next)
+                    .log("Execution with retry failed, will retry");
+            scheduledExecutorService.schedule(() ->
+                            executeWithRetry(supplier, backoff, scheduledExecutorService, callback),
+                    next, TimeUnit.MILLISECONDS);
+        }
     }
 
 }