[fix][broker] Filter system topic when getting topic list by binary proto. (#19667)

Author: shibd

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java

@@ -2182,8 +2182,8 @@ protected void handleGetTopicsOfNamespace(CommandGetTopicsOfNamespace commandGet
                     getBrokerService().pulsar().getNamespaceService().getListOfTopics(namespaceName, mode)
                         .thenAccept(topics -> {
                             boolean filterTopics = false;
-                            // filter transaction internal topic
-                            List<String> filteredTopics = TopicList.filterTransactionInternalName(topics);
+                            // filter system topic
+                            List<String> filteredTopics = TopicList.filterSystemTopic(topics);
 
                             if (enableSubscriptionPatternEvaluation && topicsPattern.isPresent()) {
                                 if (topicsPattern.get().length() <= maxSubscriptionPatternLength) {

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PatternTopicsConsumerImplTest.java

@@ -63,6 +63,8 @@ public class PatternTopicsConsumerImplTest extends ProducerConsumerBase {
     public void setup() throws Exception {
         // set isTcpLookup = true, to use BinaryProtoLookupService to get topics for a pattern.
         isTcpLookup = true;
+        // enabled transaction, to test pattern consumers not subscribe to transaction system topic.
+        conf.setTransactionCoordinatorEnabled(true);
         super.internalSetup();
         super.producerBaseSetup();
     }
@@ -251,6 +253,58 @@ public void testBinaryProtoToGetTopicsOfNamespacePersistent() throws Exception {
         producer4.close();
     }
 
+    @Test(timeOut = testTimeout)
+    public void testBinaryProtoSubscribeAllTopicOfNamespace() throws Exception {
+        String key = "testBinaryProtoSubscribeAllTopicOfNamespace";
+        String subscriptionName = "my-ex-subscription-" + key;
+        String topicName1 = "persistent://my-property/my-ns/pattern-topic-1-" + key;
+        String topicName2 = "persistent://my-property/my-ns/pattern-topic-2-" + key;
+        String topicName3 = "persistent://my-property/my-ns/pattern-topic-3-" + key;
+        Pattern pattern = Pattern.compile("my-property/my-ns/.*");
+
+        // 1. create partition
+        TenantInfoImpl tenantInfo = createDefaultTenantInfo();
+        admin.tenants().createTenant("prop", tenantInfo);
+        admin.topics().createPartitionedTopic(topicName1, 1);
+        admin.topics().createPartitionedTopic(topicName2, 2);
+        admin.topics().createPartitionedTopic(topicName3, 3);
+
+        // 2. create producer to trigger create system topic.
+        Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName1)
+                .enableBatching(false)
+                .messageRoutingMode(MessageRoutingMode.SinglePartition)
+                .create();
+
+        Consumer<byte[]> consumer = pulsarClient.newConsumer()
+                .topicsPattern(pattern)
+                .patternAutoDiscoveryPeriod(2)
+                .subscriptionName(subscriptionName)
+                .subscriptionType(SubscriptionType.Shared)
+                .ackTimeout(ackTimeOutMillis, TimeUnit.MILLISECONDS)
+                .receiverQueueSize(4)
+                .subscribe();
+        assertTrue(consumer.getTopic().startsWith(PatternMultiTopicsConsumerImpl.DUMMY_TOPIC_NAME_PREFIX));
+
+        // 4. verify consumer get methods, to get right number of partitions and topics.
+        assertSame(pattern, ((PatternMultiTopicsConsumerImpl<?>) consumer).getPattern());
+        List<String> topics = ((PatternMultiTopicsConsumerImpl<?>) consumer).getPartitions();
+        List<ConsumerImpl<byte[]>> consumers = ((PatternMultiTopicsConsumerImpl<byte[]>) consumer).getConsumers();
+
+        assertEquals(topics.size(), 6);
+        assertEquals(consumers.size(), 6);
+        assertEquals(((PatternMultiTopicsConsumerImpl<?>) consumer).getPartitionedTopics().size(), 3);
+
+        topics.forEach(topic -> log.info("topic: {}", topic));
+        consumers.forEach(c -> log.info("consumer: {}", c.getTopic()));
+
+        IntStream.range(0, topics.size()).forEach(index ->
+                assertEquals(consumers.get(index).getTopic(), topics.get(index)));
+
+        consumer.unsubscribe();
+        producer.close();
+        consumer.close();
+    }
+
     @Test(timeOut = testTimeout)
     public void testPubRateOnNonPersistent() throws Exception {
         cleanup();

pulsar-common/src/main/java/org/apache/pulsar/common/topics/TopicList.java

@@ -57,9 +57,9 @@ public static List<String> filterTopics(List<String> original, Pattern topicsPat
                 .collect(Collectors.toList());
     }
 
-    public static List<String> filterTransactionInternalName(List<String> original) {
+    public static List<String> filterSystemTopic(List<String> original) {
         return original.stream()
-                .filter(topic -> !SystemTopicNames.isTransactionInternalName(TopicName.get(topic)))
+                .filter(topic -> !SystemTopicNames.isSystemTopic(TopicName.get(topic)))
                 .collect(Collectors.toList());
     }