AdminManager: handle properly async deletePartitionedTopic

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/AdminManager.java

@@ -63,6 +63,7 @@ class AdminManager {
 
     private final PulsarAdmin admin;
     private final int defaultNumPartitions;
+
     private volatile Map<String, Set<Node>> brokersCache = Maps.newHashMap();
     private final ReentrantReadWriteLock brokersCacheLock = new ReentrantReadWriteLock();
 
@@ -212,7 +213,11 @@ CompletableFuture<Map<ConfigResource, DescribeConfigsResponse.Config>> describeC
                                 future.complete(new DescribeConfigsResponse.Config(ApiError.NONE, dummyConfig));
                                 break;
                             default:
-                                throw new InvalidRequestException("Unsupported resource type: " + resource.type());
+                                return CompletableFuture.completedFuture(new DescribeConfigsResponse.Config(
+                                        ApiError.fromThrowable(
+                                            new InvalidRequestException("Unsupported resource type: "
+                                                    + resource.type())),
+                                            Collections.emptyList()));
                         }
                         return future;
                     } catch (Exception e) {
@@ -222,6 +227,10 @@ CompletableFuture<Map<ConfigResource, DescribeConfigsResponse.Config>> describeC
                 }));
         CompletableFuture<Map<ConfigResource, DescribeConfigsResponse.Config>> resultFuture = new CompletableFuture<>();
         CompletableFuture.allOf(futureMap.values().toArray(new CompletableFuture[0])).whenComplete((ignored, e) -> {
+            if (e != null) {
+                resultFuture.completeExceptionally(e);
+                return;
+            }
             resultFuture.complete(futureMap.entrySet().stream().collect(
                     Collectors.toMap(Map.Entry::getKey, entry -> entry.getValue().getNow(null))
             ));
@@ -240,14 +249,17 @@ private DescribeConfigsResponse.ConfigEntry buildDummyEntryConfig(String configN
     public void deleteTopic(String topicToDelete,
                             Consumer<String> successConsumer,
                             Consumer<String> errorConsumer) {
-        try {
-            admin.topics().deletePartitionedTopic(topicToDelete);
-            successConsumer.accept(topicToDelete);
-            log.info("delete topic {} successfully.", topicToDelete);
-        } catch (PulsarAdminException e) {
-            log.error("delete topic {} failed, exception: ", topicToDelete, e);
-            errorConsumer.accept(topicToDelete);
-        }
+        admin.topics()
+                .deletePartitionedTopicAsync(topicToDelete)
+                .thenRun(() -> {
+                    log.info("delete topic {} successfully.", topicToDelete);
+                    successConsumer.accept(topicToDelete);
+                })
+                .exceptionally((e -> {
+                    log.error("delete topic {} failed, exception: ", topicToDelete, e);
+                    errorConsumer.accept(topicToDelete);
+                    return null;
+                }));
     }
 
     public void truncateTopic(String topicToDelete,
@@ -278,8 +290,6 @@ public void truncateTopic(String topicToDelete,
 
     }
 
-
-
     CompletableFuture<Map<String, ApiError>> createPartitionsAsync(Map<String, NewPartitions> createInfo,
                                                                    int timeoutMs,
                                                                    String namespacePrefix) {
@@ -434,4 +444,5 @@ public void setBrokers(Map<String, Set<Node>> newBrokers) {
             brokersCacheLock.writeLock().unlock();
         }
     }
+
 }