address comment

apache · May 13, 2024 · 0eb3c04 · 0eb3c04
1 parent cb9007b
commit 0eb3c04
Showing 1 changed file with 8 additions and 6 deletions.
diff --git a/...roker/src/main/java/org/apache/pulsar/broker/service/PulsarMetadataEventSynchronizer.java b/...roker/src/main/java/org/apache/pulsar/broker/service/PulsarMetadataEventSynchronizer.java
@@ -156,8 +156,8 @@ private void startProducer() {
                     State stateTransient = state;
                     log.info("[{}] Closing the new producer because the synchronizer state is {}", prod,
                             stateTransient);
-                    CompletableFuture closeProducer = new CompletableFuture<>();
-                    closeResource(() -> prod.closeAsync(), closeProducer);
+                    CompletableFuture<Void> closeProducer = new CompletableFuture<>();
+                    closeResource(prod::closeAsync, closeProducer);
                     closeProducer.thenRun(() -> {
                         log.info("[{}] Closed the new producer because the synchronizer state is {}", prod,
                                 stateTransient);
@@ -221,11 +221,13 @@ private void startConsumer() {
                 log.info("successfully created consumer {}", topicName);
             } else {
                 State stateTransient = state;
-                log.info("[{}] Closing the new consumer because the synchronizer state is {}", stateTransient);
-                CompletableFuture closeConsumer = new CompletableFuture<>();
-                closeResource(() -> consumer.closeAsync(), closeConsumer);
+                log.info("[{}] Closing the new consumer because the synchronizer state is {}", topicName,
+                        stateTransient);
+                CompletableFuture<Void> closeConsumer = new CompletableFuture<>();
+                closeResource(consumer::closeAsync, closeConsumer);
                 closeConsumer.thenRun(() -> {
-                    log.info("[{}] Closed the new consumer because the synchronizer state is {}", stateTransient);
+                    log.info("[{}] Closed the new consumer because the synchronizer state is {}", topicName,
+                            stateTransient);
                 });
             }
         }).exceptionally(ex -> {