[pulsar] 02/07: [Issue] Fix get schemaName by partitioned topic name (#7708)

[rx...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

rxl pushed a commit to branch branch-2.6
in repository https://gitbox.apache.org/repos/asf/pulsar.git

commit 84d5ab086d18812ae9ba29be602f6a347ef81564
Author: ran <ga...@126.com>
AuthorDate: Wed Aug 5 12:01:07 2020 +0800

    [Issue] Fix get schemaName by partitioned topic name (#7708)
    
    Fixes #7676
    
    ### Motivation
    
    Currently, there is an error that gets schemaName by partitioned topic name.
    
    ### Modifications
    
    Add partitioned check in method `topicName. getSchemaName()`.
    
    (cherry picked from commit 7525ab90cb5c072237a3ae2bc665930cbc691f8a)
---
 .../pulsar/schema/PartitionedTopicSchemaTest.java  | 112 +++++++++++++++++++++
 .../org/apache/pulsar/common/naming/TopicName.java |   2 +-
 2 files changed, 113 insertions(+), 1 deletion(-)

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/schema/PartitionedTopicSchemaTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/schema/PartitionedTopicSchemaTest.java
new file mode 100644
index 0000000..c112c13
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/schema/PartitionedTopicSchemaTest.java
@@ -0,0 +1,112 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.schema;
+
+import com.google.common.collect.Sets;
+import lombok.Cleanup;
+import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
+import org.apache.pulsar.client.api.schema.GenericRecord;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.policies.data.ClusterData;
+import org.apache.pulsar.common.policies.data.TenantInfo;
+import org.testng.Assert;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+/**
+ * Test get partitioned topic schema.
+ */
+public class PartitionedTopicSchemaTest extends MockedPulsarServiceBaseTest {
+
+    private final static String PARTITIONED_TOPIC = "public/default/partitioned-schema-topic";
+    private final static int MESSAGE_COUNT_PER_PARTITION  = 12;
+    private final static int TOPIC_PARTITION = 3;
+
+    @BeforeMethod
+    @Override
+    protected void setup() throws Exception {
+        isTcpLookup = true;
+        super.internalSetup();
+
+        admin.clusters().createCluster("test", new ClusterData(pulsar.getWebServiceAddress()));
+        admin.tenants().createTenant("my-property",
+                new TenantInfo(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet("test")));
+        admin.namespaces().createNamespace("my-property/my-ns");
+        admin.namespaces().setNamespaceReplicationClusters("my-property/my-ns", Sets.newHashSet("test"));
+
+        // so that clients can test short names
+        admin.tenants().createTenant("public",
+                new TenantInfo(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet("test")));
+        admin.namespaces().createNamespace("public/default");
+        admin.namespaces().setNamespaceReplicationClusters("public/default", Sets.newHashSet("test"));
+        admin.topics().createPartitionedTopic(PARTITIONED_TOPIC, TOPIC_PARTITION);
+    }
+
+    @AfterMethod
+    @Override
+    protected void cleanup() throws Exception {
+        super.internalCleanup();
+    }
+
+    @Test
+    public void test() throws Exception {
+        Consumer<GenericRecord> consumer = pulsarClient.newConsumer(Schema.AUTO_CONSUME())
+                .topic(PARTITIONED_TOPIC)
+                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+                .subscriptionName("test")
+                .subscribe();
+        consumer.close();
+
+        @Cleanup
+        Producer<Schemas.PersonFour> producer = pulsarClient.newProducer(Schema.JSON(Schemas.PersonFour.class))
+                .topic(PARTITIONED_TOPIC)
+                .enableBatching(false)
+                .roundRobinRouterBatchingPartitionSwitchFrequency(1)
+                .create();
+
+        for (int i = 0; i < MESSAGE_COUNT_PER_PARTITION * TOPIC_PARTITION; i++) {
+            Schemas.PersonFour person = new Schemas.PersonFour();
+            person.setId(i);
+            person.setName("user-" + i);
+            person.setAge(18);
+            producer.newMessage().value(person).send();
+        }
+
+        consumer = pulsarClient.newConsumer(Schema.AUTO_CONSUME())
+                .topic(TopicName.get(PARTITIONED_TOPIC).getPartition(1).toString())
+                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+                .subscriptionName("test")
+                .subscribe();
+
+        int receiveMsgCount = 0;
+        for (int i = 0; i < MESSAGE_COUNT_PER_PARTITION; i++) {
+            Message<GenericRecord> message = consumer.receive();
+            Assert.assertNotNull(message);
+            receiveMsgCount++;
+        }
+        Assert.assertEquals(MESSAGE_COUNT_PER_PARTITION, receiveMsgCount);
+    }
+
+}
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java b/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java
index 5ec0d58..49eec35 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java
@@ -342,7 +342,7 @@ public class TopicName implements ServiceUnitId {
     public String getSchemaName() {
         return getTenant()
             + "/" + getNamespacePortion()
-            + "/" + getEncodedLocalName();
+            + "/" + TopicName.get(getPartitionedTopicName()).getEncodedLocalName();
     }
 
     @Override