clean up

Author: eolivelli

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/NonPersistentTopics.java

@@ -128,7 +128,13 @@ public StreamingOutput getInternalStats(
             @QueryParam("metadata") @DefaultValue("false") boolean metadata) {
         validateTopicName(tenant, namespace, encodedTopic);
         return output -> {
-            internalGetInternalStatsAsync(authoritative, metadata)
+            validateTopicOwnershipAsync(topicName, authoritative)
+                    .thenCompose(__ -> validateTopicOperationAsync(topicName, TopicOperation.GET_STATS))
+                    .thenCompose(__ -> {
+                        Topic topic = getTopicReference(topicName);
+                        boolean includeMetadata = metadata && hasSuperUserAccess();
+                        return topic.getInternalStats(includeMetadata);
+                    })
                     .thenAccept(stats -> {
                         try {
                             ObjectMapperFactory.getMapper().getObjectMapper().writeValue(output, stats);

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java

@@ -1407,7 +1407,7 @@ public void deleteSubscription(
         internalDeleteSubscriptionAsync(subName, authoritative, force)
                 .thenRun(() -> asyncResponse.resume(Response.noContent().build()))
                 .exceptionally(ex -> {
-                    Throwable cause = unwrapCompletionException(ex);
+                    Throwable cause = FutureUtil.unwrapCompletionException(ex);
 
                     // If the exception is not redirect exception we need to log it.
                     if (!isRedirectException(cause)) {
@@ -1699,7 +1699,7 @@ public void resetCursor(
         internalResetCursorAsync(decode(encodedSubName), timestamp, authoritative)
             .thenAccept(__ -> asyncResponse.resume(Response.noContent().build()))
             .exceptionally(ex -> {
-                Throwable t = unwrapCompletionException(ex);
+                Throwable t = FutureUtil.unwrapCompletionException(ex);
                 if (!isRedirectException(t)) {
                     log.error("[{}][{}] Failed to reset cursor on subscription {} to time {}",
                         clientAppId(), topicName, encodedSubName, timestamp, t);
@@ -2065,7 +2065,7 @@ public void getBacklog(
                 .thenCompose(__ ->  internalGetBacklogAsync(authoritative))
                 .thenAccept(asyncResponse::resume)
                 .exceptionally(ex -> {
-                    Throwable t = unwrapCompletionException(ex);
+                    Throwable t = FutureUtil.unwrapCompletionException(ex);
                     if (t instanceof MetadataStoreException.NotFoundException) {
                         log.warn("[{}] Failed to get topic backlog {}: Namespace does not exist", clientAppId(),
                                 namespaceName);
@@ -4219,7 +4219,7 @@ public void truncateTopic(
         internalTruncateTopicAsync(authoritative)
             .thenAccept(__ -> asyncResponse.resume(Response.noContent().build()))
             .exceptionally(ex -> {
-                Throwable t = unwrapCompletionException(ex);
+                Throwable t = FutureUtil.unwrapCompletionException(ex);
                 if (!isRedirectException(t)) {
                     log.error("[{}] Failed to truncate topic {}", clientAppId(), topicName, t);
                 }

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

@@ -2387,6 +2387,7 @@ public CompletableFuture<SchemaVersion> deleteSchema() {
 
     @Override
     public CompletableFuture<PersistentTopicInternalStats> getInternalStats(boolean includeLedgerMetadata) {
+
         CompletableFuture<PersistentTopicInternalStats> statFuture = new CompletableFuture<>();
         PersistentTopicInternalStats stats = new PersistentTopicInternalStats();