private CacheZone parseCacheZone(
            List<SortSourceStreamSinkInfo> sinks,
            SortSourceClusterInfo cluster,
            boolean isBackupTag) {
        switch (cluster.getType()) {
            case ClusterType.PULSAR:
                return parsePulsarZone(sinks, cluster, isBackupTag);
            default:
                throw new BusinessException(String.format("do not support cluster type=%s of cluster=%s",
                        cluster.getType(), cluster));
        }
    }
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