private Map<String, CacheZone> parseCacheZones(
            String clusterName,
            List<SortSourceStreamSinkInfo> sinkList) {

        Preconditions.expectNotNull(sortClusters.get(clusterName), "sort cluster should not be NULL");
        Set<String> tags = sortClusters.get(clusterName).getClusterTagsSet();

        // get group infos
        List<SortSourceStreamSinkInfo> sinkInfoList = sinkList.stream()
                .filter(sinkInfo -> groupInfos.containsKey(sinkInfo.getGroupId())
                        && allStreams.containsKey(sinkInfo.getGroupId())
                        && allStreams.get(sinkInfo.getGroupId()).containsKey(sinkInfo.getStreamId()))
                .collect(Collectors.toList());

        // group them by cluster tag.
        Map<String, List<SortSourceStreamSinkInfo>> tag2SinkInfos = sinkInfoList.stream()
                .filter(sink -> Objects.nonNull(groupInfos.get(sink.getGroupId())))
                .filter(sink -> {
                    if (CollectionUtils.isEmpty(tags)) {
                        return true;
                    }
                    return tags.contains(groupInfos.get(sink.getGroupId()).getClusterTag());
                })
                .collect(Collectors.groupingBy(sink -> {
                    SortSourceGroupInfo groupInfo = groupInfos.get(sink.getGroupId());
                    return groupInfo.getClusterTag();
                }));

        // group them by second cluster tag.
        Map<String, List<SortSourceStreamSinkInfo>> backupTag2SinkInfos = sinkInfoList.stream()
                .filter(sink -> backupClusterTag.containsKey(sink.getGroupId()))
                .filter(sink -> {
                    if (CollectionUtils.isEmpty(tags)) {
                        return true;
                    }
                    return tags.contains(backupClusterTag.get(sink.getGroupId()));
                })
                .collect(Collectors.groupingBy(info -> backupClusterTag.get(info.getGroupId())));

        List<CacheZone> cacheZones = this.parseCacheZonesByTag(tag2SinkInfos, false);
        List<CacheZone> backupCacheZones = this.parseCacheZonesByTag(backupTag2SinkInfos, true);

        return Stream.of(cacheZones, backupCacheZones)
                .flatMap(Collection::stream)
                .collect(Collectors.toMap(
                        CacheZone::getZoneName,
                        cacheZone -> cacheZone,
                        (zone1, zone2) -> {
                            zone1.getTopics().addAll(zone2.getTopics());
                            return zone1;
                        }));
    }
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