private void reloadAllConfigs() {

        // reload mq cluster and sort cluster
        List<SortSourceClusterInfo> allClusters = configLoader.loadAllClusters();
        sortClusters = allClusters.stream()
                .collect(Collectors.toMap(SortSourceClusterInfo::getName, v -> v));

        // group mq clusters by cluster tag
        mqClusters = new HashMap<>();
        allClusters.stream()
                .filter(cluster -> SUPPORTED_MQ_TYPE.contains(cluster.getType()))
                .filter(SortSourceClusterInfo::isConsumable)
                .forEach(mq -> {
                    Set<String> tags = mq.getClusterTagsSet();
                    tags.forEach(tag -> {
                        List<SortSourceClusterInfo> list = mqClusters.computeIfAbsent(tag, k -> new ArrayList<>());
                        list.add(mq);
                    });
                });

        // reload all stream sinks, to Map<clusterName, Map<taskName, List<groupId>>> format
        List<SortSourceStreamSinkInfo> allStreamSinks = configLoader.loadAllStreamSinks();
        streamSinkMap = new HashMap<>();
        allStreamSinks.stream()
                .filter(sink -> StringUtils.isNotBlank(sink.getSortClusterName()))
                .filter(sink -> Objects.nonNull(sortClusters.get(sink.getSortClusterName())))
                .filter(sink -> StringUtils.isNotBlank(sink.getSortTaskName()))
                .forEach(sink -> {
                    Map<String, List<SortSourceStreamSinkInfo>> task2groupsMap =
                            streamSinkMap.computeIfAbsent(sink.getSortClusterName(), k -> new ConcurrentHashMap<>());
                    List<SortSourceStreamSinkInfo> sinkInfoList =
                            task2groupsMap.computeIfAbsent(sink.getSortTaskName(), k -> new ArrayList<>());
                    sinkInfoList.add(sink);
                });

        // reload all groups
        groupInfos = configLoader.loadAllGroup()
                .stream()
                .filter(group -> StringUtils.isNotBlank(group.getMqResource()))
                .filter(group -> StringUtils.isNotBlank(group.getClusterTag()))
                .collect(Collectors.toMap(SortSourceGroupInfo::getGroupId, info -> info));

        // reload all back up cluster
        backupClusterTag = configLoader.loadGroupBackupInfo(ClusterSwitch.BACKUP_CLUSTER_TAG)
                .stream()
                .collect(Collectors.toMap(InlongGroupExtEntity::getInlongGroupId, InlongGroupExtEntity::getKeyValue));

        // reload all back up group mq resource
        backupGroupMqResource = configLoader.loadGroupBackupInfo(ClusterSwitch.BACKUP_MQ_RESOURCE)
                .stream()
                .collect(Collectors.toMap(InlongGroupExtEntity::getInlongGroupId, InlongGroupExtEntity::getKeyValue));

        // reload all streams
        allStreams = configLoader.loadAllStreams()
                .stream()
                .filter(stream -> StringUtils.isNotBlank(stream.getMqResource()))
                .collect(Collectors.groupingBy(SortSourceStreamInfo::getInlongGroupId,
                        Collectors.toMap(SortSourceStreamInfo::getInlongStreamId, info -> info)));

        // reload all back up stream mq resource
        backupStreamMqResource = configLoader.loadStreamBackupInfo(ClusterSwitch.BACKUP_MQ_RESOURCE)
                .stream()
                .collect(Collectors.groupingBy(InlongStreamExtEntity::getInlongGroupId,
                        Collectors.toMap(InlongStreamExtEntity::getInlongStreamId,
                                InlongStreamExtEntity::getKeyValue)));
    }
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