Fix list non-persistent topics shows the persistent topics (#5502)

* Fix list non-persistent topics shows the persistent topics
---

Fixes #5414

*Motivation*

When using the REST API to request to list all the non-persistent
topics, it will show the persistent topics.

*Modifications*

- Add a filter when before sending the response

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

@@ -29,6 +29,7 @@ import io.swagger.annotations.ApiResponses;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
 
 import javax.ws.rs.DefaultValue;
 import javax.ws.rs.Encoded;
@@ -292,7 +293,12 @@ public class NonPersistentTopics extends PersistentTopics {
                     return null;
                 }
             }
-            asyncResponse.resume(topics);
+
+            final List<String> nonPersistentTopics =
+                topics.stream()
+                      .filter(name -> !TopicName.get(name).isPersistent())
+                      .collect(Collectors.toList());
+            asyncResponse.resume(nonPersistentTopics);
             return null;
         });
     }

tests/integration/src/test/java/org/apache/pulsar/tests/integration/cli/CLITest.java

@@ -18,6 +18,11 @@
  */
 package org.apache.pulsar.tests.integration.cli;
 
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClients;
+import org.apache.http.util.EntityUtils;
 import org.apache.pulsar.client.api.Consumer;
 import org.apache.pulsar.client.api.Message;
 import org.apache.pulsar.client.api.Producer;
@@ -304,4 +309,16 @@ public class CLITest extends PulsarTestSuite {
         client.close();
     }
 
+    @Test
+    public void testListNonPersistentTopicsCmd() throws Exception {
+        String persistentTopic = "test-list-non-persistent-topic";
+        ContainerExecResult result = pulsarCluster.runAdminCommandOnAnyBroker("topics", "create", persistentTopic);
+        assertEquals(result.getExitCode(), 0);
+        HttpGet get = new HttpGet(pulsarCluster.getHttpServiceUrl() + "/admin/v2/non-persistent/public/default");
+        try (CloseableHttpClient client = HttpClients.createDefault();
+             CloseableHttpResponse response = client.execute(get)) {
+            assertFalse(EntityUtils.toString(response.getEntity()).contains(persistentTopic));
+        }
+    }
+
 }