[fix][misc] Topic name from persistence name should decode local name

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);

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);
     }