Tolerate runtime exceptions in flushAsync

apache · Dec 16, 2024 · b462241 · b462241
1 parent 6e9cecd
commit b462241
Showing 1 changed file with 1 addition and 1 deletion.
diff --git a/...-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ProducerCache.java b/...-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ProducerCache.java
@@ -67,7 +67,7 @@ public ProducerCache() {
                 .<ProducerCacheKey, Producer<?>>removalListener((key, producer, cause) -> {
                     log.info("Closing producer for topic {}, cause {}", key.topic(), cause);
                     CompletableFuture closeFuture =
-                            producer.flushAsync()
+                            CompletableFuture.supplyAsync(() -> producer.flushAsync(), Runnable::run)
                                     .orTimeout(FLUSH_OR_CLOSE_TIMEOUT_SECONDS, TimeUnit.SECONDS)
                                     .exceptionally(ex -> {
                                         Throwable unwrappedCause = FutureUtil.unwrapCompletionException(ex);