test

pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java

@@ -684,7 +684,6 @@ public CompletableFuture<InputStream> streamInternalStatsAsync(String topic, boo
         TopicName tn = validateTopic(topic);
         WebTarget path = topicPath(tn, "internalStats");
         path = path.queryParam("metadata", metadata);
-        CompletableFuture<InputStream> future = new CompletableFuture<>();
         return asyncGetRequest(path, new FutureCallback<InputStream>() {});
     }
 

pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/PulsarAdminToolTest.java

@@ -38,8 +38,11 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Sets;
+
+import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.File;
+import java.io.InputStream;
 import java.io.PrintStream;
 import java.io.PrintWriter;
 import java.net.URL;
@@ -1560,6 +1563,11 @@ public void topics() throws Exception {
         Lookup mockLookup = mock(Lookup.class);
         when(admin.lookups()).thenReturn(mockLookup);
 
+        String statsString = "{}";
+        InputStream stream = new ByteArrayInputStream(statsString.getBytes(StandardCharsets.UTF_8));
+        when(mockTopics.streamInternalStats("persistent://myprop/clust/ns1/ds1", false)).thenReturn(stream);
+        when(mockTopics.streamPartitionedInternalStats("persistent://myprop/clust/ns1/ds1")).thenReturn(stream);
+
         CmdTopics cmdTopics = new CmdTopics(() -> admin);
 
         cmdTopics.run(split("truncate persistent://myprop/clust/ns1/ds1"));
@@ -1607,6 +1615,9 @@ public void topics() throws Exception {
         cmdTopics.run(split("stats-internal persistent://myprop/clust/ns1/ds1"));
         verify(mockTopics).getInternalStats("persistent://myprop/clust/ns1/ds1", false);
 
+        cmdTopics.run(split("stats-internal persistent://myprop/clust/ns1/ds1 --streaming"));
+        verify(mockTopics).streamInternalStats("persistent://myprop/clust/ns1/ds1", false);
+
         cmdTopics.run(split("get-backlog-quotas persistent://myprop/clust/ns1/ds1 -ap"));
         verify(mockTopics).getBacklogQuotaMap("persistent://myprop/clust/ns1/ds1", true);
         cmdTopics.run(split("set-backlog-quota persistent://myprop/clust/ns1/ds1 -l 10 -p producer_request_hold"));
@@ -1655,6 +1666,9 @@ public void topics() throws Exception {
         cmdTopics.run(split("partitioned-stats-internal persistent://myprop/clust/ns1/ds1"));
         verify(mockTopics).getPartitionedInternalStats("persistent://myprop/clust/ns1/ds1");
 
+        cmdTopics.run(split("partitioned-stats-internal persistent://myprop/clust/ns1/ds1 --streaming"));
+        verify(mockTopics).streamPartitionedInternalStats("persistent://myprop/clust/ns1/ds1");
+
         cmdTopics.run(split("clear-backlog persistent://myprop/clust/ns1/ds1 -s sub1"));
         verify(mockTopics).skipAllMessages("persistent://myprop/clust/ns1/ds1", "sub1");
 

pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java

@@ -833,7 +833,7 @@ private class GetPartitionedStatsInternal extends CliCommand {
         void run() throws Exception {
             String topic = validateTopicName(topicName);
             if (streaming) {
-                try (InputStream in = getTopics().streamPartitionedInternalStats(topic);) {
+                try (InputStream in = getTopics().streamPartitionedInternalStats(topic)) {
                     int size;
                     byte[] buffer = new byte[2048];
                     while ((size = in.read(buffer)) != -1) {