@Override
    public SortSourceConfigResponse getSourceConfig(
            String cluster,
            String task,
            String md5) {

        // if cluster or task are invalid
        if (StringUtils.isBlank(cluster) || StringUtils.isBlank(task)) {
            String errMsg = "blank cluster name or task name, return nothing";
            LOGGER.debug(errMsg);
            return SortSourceConfigResponse.builder()
                    .code(RESPONSE_CODE_REQ_PARAMS_ERROR)
                    .msg(errMsg)
                    .build();
        }

        // if there is no config, but still return success
        if (!sortSourceConfigMap.containsKey(cluster) || !sortSourceConfigMap.get(cluster).containsKey(task)) {
            String errMsg = String.format("there is no valid source config of cluster %s, task %s", cluster, task);
            LOGGER.debug(errMsg);
            return SortSourceConfigResponse.builder()
                    .code(RESPONSE_CODE_SUCCESS)
                    .msg(errMsg)
                    .build();
        }

        // if the same md5
        if (sortSourceMd5Map.get(cluster).get(task).equals(md5)) {
            return SortSourceConfigResponse.builder()
                    .code(RESPONSE_CODE_NO_UPDATE)
                    .msg("No update")
                    .md5(md5)
                    .build();
        }

        // if there is bad config
        if (sortSourceConfigMap.get(cluster).get(task).getCacheZones().isEmpty()) {
            String errMsg = String.format("find empty cache zones of cluster %s, task %s, "
                    + "please check the manager log", cluster, task);
            LOGGER.debug(errMsg);
            return SortSourceConfigResponse.builder()
                    .code(RESPONSE_CODE_FAIL)
                    .msg(errMsg)
                    .build();
        }

        return SortSourceConfigResponse.builder()
                .code(RESPONSE_CODE_SUCCESS)
                .msg("Success")
                .data(sortSourceConfigMap.get(cluster).get(task))
                .md5(sortSourceMd5Map.get(cluster).get(task))
                .build();

    }
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