private static void appendSources(
            Map<String, String> flumeConf,
            String name, Map<String, String> sinkParams) {
        // sources
        String sourceName = name + "Source";
        flumeConf.put(name + ".sources", sourceName);
        StringBuilder builder = new StringBuilder();
        String prefix = builder.append(name).append(".sources.").append(sourceName).append(".").toString();
        // type
        builder.setLength(0);
        String sourceType = builder.append(prefix).append("type").toString();
        String sourceClass = sinkParams.getOrDefault(KEY_SORT_SOURCE_TYPE,
                CommonPropertiesHolder.getString(KEY_SORT_SOURCE_TYPE));
        flumeConf.put(sourceType, sourceClass);
        // channel
        builder.setLength(0);
        String channelKey = builder.append(prefix).append("channels").toString();
        String channelName = name + "Channel";
        flumeConf.put(channelKey, channelName);
        // selector.type
        builder.setLength(0);
        String selectorTypeKey = builder.append(prefix).append("selector.type").toString();
        flumeConf.put(selectorTypeKey, "org.apache.flume.channel.ReplicatingChannelSelector");
        // valid msg time interval
        builder.setLength(0);
        String interceptorKey = builder.append(prefix).append("interceptors").toString();
        String interceptorName = name + "Interceptor";
        flumeConf.put(interceptorKey, interceptorName);

        builder.setLength(0);
        String interceptorType = builder.append(prefix).append("interceptors.").append(interceptorName)
                .append(".type").toString();
        flumeConf.put(interceptorType,
                CommonPropertiesHolder.getString(KEY_SORT_INTERCEPTOR_TYPE, DEFAULT_SORT_INTERCEPTOR_TYPE));
        builder.setLength(0);
        String startTimeKey = builder.append(prefix).append("interceptors.").append(interceptorName).append(".")
                .append(KEY_ROLLBACK_START_TIME).toString();
        Optional.ofNullable(CommonPropertiesHolder.getString(KEY_ROLLBACK_START_TIME))
                .map(startTime -> flumeConf.put(startTimeKey, startTime));
        builder.setLength(0);
        String stopTimeKey = builder.append(prefix).append("interceptors.").append(interceptorName).append(".")
                .append(KEY_ROLLBACK_STOP_TIME).toString();
        Optional.ofNullable(CommonPropertiesHolder.getString(KEY_ROLLBACK_STOP_TIME))
                .map(stopTime -> flumeConf.put(stopTimeKey, stopTime));

        appendCommon(flumeConf, prefix, null, name);
    }
append source config

@param flumeConf final config of flume
@param name sort task name
@param sinkParams sink params of this task
inlong-sort-standalone/sort-standalone-common/src/main/java/org/apache/inlong/sort/standalone/utils/FlumeConfigGenerator.java