private void reloadAllClusterConfig() {
        // load all fields info
        List<SortFieldInfo> fieldInfos = configLoader.loadAllFields();
        fieldMap = new HashMap<>();
        fieldInfos.forEach(info -> {
            List<String> fields = fieldMap.computeIfAbsent(info.getSinkId(), k -> new ArrayList<>());
            fields.add(info.getFieldName());
        });

        List<StreamSinkEntity> sinkEntities = configLoader.loadAllStreamSinkEntity();
        // get all task under a given cluster, has been reduced into cluster and task.
        List<SortTaskInfo> tasks = configLoader.loadAllTask();
        Map<String, List<SortTaskInfo>> clusterTaskMap = tasks.stream()
                .filter(dto -> StringUtils.isNotBlank(dto.getSortClusterName())
                        && StringUtils.isNotBlank(dto.getSortTaskName())
                        && StringUtils.isNotBlank(dto.getDataNodeName())
                        && StringUtils.isNotBlank(dto.getSinkType()))
                .collect(Collectors.groupingBy(SortTaskInfo::getSortClusterName));

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

        // get all stream sinks
        Map<String, List<StreamSinkEntity>> task2AllStreams = sinkEntities.stream()
                .filter(entity -> StringUtils.isNotBlank(entity.getInlongClusterName()))
                .filter(entity -> StringUtils.isNotBlank(entity.getSortTaskName()))
                .filter(entity -> StringUtils.isNotBlank(entity.getDataNodeName()))
                .collect(Collectors.groupingBy(StreamSinkEntity::getSortTaskName));

        // get all data nodes and group by node name
        List<DataNodeEntity> dataNodeEntities = configLoader.loadAllDataNodeEntity();
        Map<String, DataNodeInfo> task2DataNodeMap = dataNodeEntities.stream()
                .filter(entity -> StringUtils.isNotBlank(entity.getName()))
                .map(entity -> {
                    DataNodeOperator operator = dataNodeOperatorFactory.getInstance(entity.getType());
                    return operator.getFromEntity(entity);
                })
                .collect(Collectors.toMap(DataNodeInfo::getName, info -> info));

        // re-org all SortClusterConfigs
        Map<String, SortClusterConfig> newConfigMap = new ConcurrentHashMap<>();
        Map<String, String> newMd5Map = new ConcurrentHashMap<>();
        Map<String, String> newErrorLogMap = new ConcurrentHashMap<>();

        clusterTaskMap.forEach((clusterName, taskList) -> {
            try {
                SortClusterConfig config = this.getConfigByClusterName(clusterName,
                        taskList, task2AllStreams, task2DataNodeMap);
                String jsonStr = GSON.toJson(config);
                String md5 = DigestUtils.md5Hex(jsonStr);
                newConfigMap.put(clusterName, config);
                newMd5Map.put(clusterName, md5);
            } catch (Throwable e) {
                // if get config failed, update the err log.
                String errMsg = Optional.ofNullable(e.getMessage()).orElse("Unknown error, please check logs");
                newErrorLogMap.put(clusterName, errMsg);
                LOGGER.error("Failed to update cluster config={}", clusterName, e);
            }
        });

        sortClusterErrorLogMap = newErrorLogMap;
        sortClusterConfigMap = newConfigMap;
        sortClusterMd5Map = newMd5Map;
    }
Used to cache the sort cluster config and reduce the number of query to database.
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/SortClusterServiceImpl.java