From dae7d8bcf3ed701ba0c77ffb0be221757418d278 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E8=90=A7=E6=98=93=E5=AE=A2?= <kmter@live.com>
Date: Tue, 11 Jun 2024 12:46:04 +0800
Subject: [PATCH] [fix][misc] Topic name from persistence name should decode
 local name (#22879)

(cherry picked from commit c326d8e2203b6e9be37f4f2066fd7e90a9b9fb54)
---
 .../java/org/apache/pulsar/common/naming/TopicName.java     | 5 ++---
 .../java/org/apache/pulsar/common/naming/TopicNameTest.java | 6 ++++++
 2 files changed, 8 insertions(+), 3 deletions(-)

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 69d3aaeeebb4d..5ef8f15b800f7 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
@@ -358,17 +358,16 @@ public static String fromPersistenceNamingEncoding(String mlName) {
         String localName;
         if (parts.size() == 4) {
             tenant = parts.get(0);
-            cluster = null;
             namespacePortion = parts.get(1);
             domain = parts.get(2);
-            localName = parts.get(3);
+            localName = Codec.decode(parts.get(3));
             return String.format("%s://%s/%s/%s", domain, tenant, namespacePortion, localName);
         } else if (parts.size() == 5) {
             tenant = parts.get(0);
             cluster = parts.get(1);
             namespacePortion = parts.get(2);
             domain = parts.get(3);
-            localName = parts.get(4);
+            localName = Codec.decode(parts.get(4));
             return String.format("%s://%s/%s/%s/%s", domain, tenant, cluster, namespacePortion, localName);
         } else {
             throw new IllegalArgumentException("Invalid managedLedger name: " + mlName);
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/naming/TopicNameTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/naming/TopicNameTest.java
index 6d3ef7599d30e..abd9d5b427848 100644
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/naming/TopicNameTest.java
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/naming/TopicNameTest.java
@@ -267,6 +267,12 @@ public void testFromPersistenceNamingEncoding() {
         } catch (IllegalArgumentException e) {
             // Exception is expected.
         }
+
+        // case5: local name with special characters e.g. a:b:c
+        String topicName = "persistent://tenant/namespace/a:b:c";
+        String persistentNamingEncoding = "tenant/namespace/persistent/a%3Ab%3Ac";
+        assertEquals(TopicName.get(topicName).getPersistenceNamingEncoding(), persistentNamingEncoding);
+        assertEquals(TopicName.fromPersistenceNamingEncoding(persistentNamingEncoding), topicName);
     }