[Broker] Fix call sync method in async rest api for internalTerminatePartitionedTopic
@@ -3313,38 +3313,45 @@ public class PersistentTopicsBase extends AdminResource {
}
protected void internalTerminatePartitionedTopic(AsyncResponse asyncResponse, boolean authoritative) {
+ CompletableFuture<Void> future;
if (topicName.isGlobal()) {
- validateGlobalNamespaceOwnership(namespaceName);
+ future = validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ future = CompletableFuture.completedFuture(null);
}
- validateTopicOwnership(topicName, authoritative);
- validateTopicOperation(topicName, TopicOperation.TERMINATE);
-
- Map<Integer, MessageId> messageIds = new ConcurrentHashMap<>();
-
- PartitionedTopicMetadata partitionMetadata = getPartitionedTopicMetadata(topicName, authoritative, false);
+ future.thenCompose(__ -> validateTopicOperationAsync(topicName, TopicOperation.TERMINATE)
+ .thenCompose(unused -> getPartitionedTopicMetadataAsync(topicName, authoritative, false))
+ .thenAccept(partitionMetadata -> {
if (partitionMetadata.partitions == 0) {
- throw new RestException(Status.METHOD_NOT_ALLOWED, "Termination of a non-partitioned topic is "
- + "not allowed using partitioned-terminate, please use terminate commands.");
+ String msg = "Termination of a non-partitioned topic is not allowed using partitioned-terminate"
+ + ", please use terminate commands";
+ log.error("[{}] [{}] {}", clientAppId(), topicName, msg);
+ asyncResponse.resume(new RestException(Status.METHOD_NOT_ALLOWED, msg));
+ return;
}
if (partitionMetadata.partitions > 0) {
- final List<CompletableFuture<MessageId>> futures = Lists.newArrayList();
+ Map<Integer, MessageId> messageIds = new ConcurrentHashMap<>(partitionMetadata.partitions);
+ final List<CompletableFuture<MessageId>> futures =
+ Lists.newArrayListWithCapacity(partitionMetadata.partitions);
for (int i = 0; i < partitionMetadata.partitions; i++) {
TopicName topicNamePartition = topicName.getPartition(i);
try {
int finalI = i;
futures.add(pulsar().getAdminClient().topics()
- .terminateTopicAsync(topicNamePartition.toString()).whenComplete((messageId, throwable) -> {
+ .terminateTopicAsync(topicNamePartition.toString())
+ .whenComplete((messageId, throwable) -> {
if (throwable != null) {
- log.error("[{}] Failed to terminate topic {}", clientAppId(), topicNamePartition,
- throwable);
+ log.error("[{}] Failed to terminate topic {}", clientAppId(),
+ topicNamePartition, throwable);
asyncResponse.resume(new RestException(throwable));
}
messageIds.put(finalI, messageId);
}));
} catch (Exception e) {
- log.error("[{}] Failed to terminate topic {}", clientAppId(), topicNamePartition, e);
+ log.error("[{}] Failed to terminate topic {}", clientAppId(), topicNamePartition,
+ e);
throw new RestException(e);
}
}
@@ -3362,6 +3369,18 @@ public class PersistentTopicsBase extends AdminResource {
return null;
});
}
+ }).exceptionally(e -> {
+ Throwable cause = e.getCause();
+ log.error("[{}] Failed to terminate topic {}", clientAppId(), topicName, cause);
+ resumeAsyncResponseExceptionally(asyncResponse, cause);
+ return null;
+ })
+ ).exceptionally(e -> {
+ Throwable cause = e.getCause();
+ log.error("[{}] Failed to terminate topic {}", clientAppId(), topicName, cause);
+ resumeAsyncResponseExceptionally(asyncResponse, cause);
+ return null;
+ });
}
protected void internalExpireMessagesByTimestamp(AsyncResponse asyncResponse, String subName,
