private List<CacheZone> parseCacheZonesByTag(
            Map<String, List<SortSourceStreamSinkInfo>> tag2Sinks,
            boolean isBackup) {

        return tag2Sinks.keySet().stream()
                .filter(mqClusters::containsKey)
                .flatMap(tag -> {
                    List<SortSourceStreamSinkInfo> sinks = tag2Sinks.get(tag);
                    List<SortSourceClusterInfo> clusters = mqClusters.get(tag);
                    return clusters.stream()
                            .map(cluster -> {
                                CacheZone zone = null;
                                try {
                                    zone = this.parseCacheZone(sinks, cluster, isBackup);
                                } catch (IllegalStateException e) {
                                    LOGGER.error("fail to init cache zone for cluster " + cluster, e);
                                }
                                return zone;
                            });
                })
                .filter(Objects::nonNull)
                .collect(Collectors.toList());
    }
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