private CacheZone parsePulsarZone(
            List<SortSourceStreamSinkInfo> sinks,
            SortSourceClusterInfo cluster,
            boolean isBackupTag) {
        Map<String, String> param = cluster.getExtParamsMap();
        String tenant = Optional.ofNullable(param.get(KEY_NEW_TENANT)).orElse(param.get(KEY_OLD_TENANT));
        String auth = param.getOrDefault(KEY_AUTH, StringUtils.EMPTY);
        List<Topic> sdkTopics = sinks.stream()
                .map(sink -> {
                    String groupId = sink.getGroupId();
                    String streamId = sink.getStreamId();
                    SortSourceGroupInfo groupInfo = groupInfos.get(groupId);
                    SortSourceStreamInfo streamInfo = allStreams.get(groupId).get(streamId);
                    try {
                        String namespace = groupInfo.getMqResource();
                        String topic = streamInfo.getMqResource();
                        if (isBackupTag) {
                            if (backupGroupMqResource.containsKey(groupId)) {
                                namespace = backupGroupMqResource.get(groupId);
                            }
                            if (backupStreamMqResource.containsKey(groupId)
                                    && backupStreamMqResource.get(groupId).containsKey(streamId)) {
                                topic = backupStreamMqResource.get(groupId).get(streamId);
                            }
                        }
                        String fullTopic = tenant + InlongConstants.SLASH + namespace + InlongConstants.SLASH + topic;

                        Map<String, String> groupExt = groupInfo.getExtParamsMap();
                        String groupTenant = Optional
                                .ofNullable(groupExt.get(KEY_NEW_TENANT))
                                .orElse(groupExt.get(KEY_OLD_TENANT));
                        if (StringUtils.isNotBlank(groupTenant)) {
                            fullTopic = groupTenant + InlongConstants.SLASH + namespace + InlongConstants.SLASH + topic;
                        }

                        return Topic.builder()
                                .topic(fullTopic)
                                .topicProperties(sink.getExtParamsMap())
                                .build();
                    } catch (Exception e) {
                        LOGGER.error("fail to parse topic of groupId={}, streamId={}", groupId, streamId, e);
                        return null;
                    }
                })
                .filter(Objects::nonNull)
                .collect(Collectors.toList());
        return CacheZone.builder()
                .zoneName(cluster.getName())
                .serviceUrl(cluster.getUrl())
                .topics(sdkTopics)
                .authentication(auth)
                .cacheZoneProperties(cluster.getExtParamsMap())
                .zoneType(ClusterType.PULSAR)
                .build();
    }
key 1: cluster name, key 2: task name, value : source config
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/SortSourceServiceImpl.java