public static String fromPersistenceNamingEncoding(String mlName) {
        // The managedLedgerName convention is: tenant/namespace/domain/topic
        // We want to transform to topic full name in the order: domain://tenant/namespace/topic
        if (mlName == null || mlName.length() == 0) {
            return mlName;
        }
        List<String> parts = Splitter.on("/").splitToList(mlName);
        String tenant;
        String cluster;
        String namespacePortion;
        String domain;
        String localName;
        if (parts.size() == 4) {
            tenant = parts.get(0);
            namespacePortion = parts.get(1);
            domain = parts.get(2);
            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 = 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);
        }
    }
@Test
    public void testFromPersistenceNamingEncoding() {
        // case1: V2
        String mlName1 = "public_tenant/default_namespace/persistent/test_topic";
        String expectedTopicName1 = "persistent://public_tenant/default_namespace/test_topic";

        TopicName name1 = TopicName.get(expectedTopicName1);
        assertEquals(name1.getPersistenceNamingEncoding(), mlName1);
        assertEquals(TopicName.fromPersistenceNamingEncoding(mlName1), expectedTopicName1);

        // case2: V1
        String mlName2 = "public_tenant/my_cluster/default_namespace/persistent/test_topic";
        String expectedTopicName2 = "persistent://public_tenant/my_cluster/default_namespace/test_topic";

        TopicName name2 = TopicName.get(expectedTopicName2);
        assertEquals(name2.getPersistenceNamingEncoding(), mlName2);
        assertEquals(TopicName.fromPersistenceNamingEncoding(mlName2), expectedTopicName2);

        // case3: null
        String mlName3 = "";
        String expectedTopicName3 = "";
        assertEquals(expectedTopicName3, TopicName.fromPersistenceNamingEncoding(mlName3));

        // case4: Invalid name
        try {
            String mlName4 = "public_tenant/my_cluster/default_namespace/persistent/test_topic/sub_topic";
            TopicName.fromPersistenceNamingEncoding(mlName4);
            fail("Should have raised exception");
        } 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);
    }