public static FlinkConfig getFlinkConfigFromFile() throws Exception {
        Path pluginPath = Paths.get(DEFAULT_PLUGINS).toAbsolutePath();
        String defaultConfigFilePath = pluginPath + File.separator + DEFAULT_CONFIG_FILE;

        log.info("Start to load Flink config from file: {}", defaultConfigFilePath);

        Properties properties = new Properties();
        try (BufferedReader bufferedReader = new BufferedReader(new FileReader(defaultConfigFilePath))) {
            properties.load(bufferedReader);
        }
        FlinkConfig flinkConfig = new FlinkConfig();
        flinkConfig.setPort(Integer.valueOf(properties.getProperty(PORT)));
        flinkConfig.setAddress(properties.getProperty(ADDRESS));
        flinkConfig.setParallelism(Integer.valueOf(properties.getProperty(PARALLELISM)));
        flinkConfig.setSavepointDirectory(properties.getProperty(SAVEPOINT_DIRECTORY));
        flinkConfig.setJobManagerPort(Integer.valueOf(properties.getProperty(JOB_MANAGER_PORT)));
        flinkConfig.setDrain(Boolean.parseBoolean(properties.getProperty(DRAIN)));
        flinkConfig.setVersion(properties.getProperty(FLINK_VERSION));
        flinkConfig.setDynamicParallelismEnable(Boolean.parseBoolean(properties.getProperty(
                Constants.FLINK_DYNAMIC_PARALLELISM_ENABLE)));
        flinkConfig.setMaxMsgRatePerCore(Integer.valueOf(properties.getProperty(Constants.FLINK_MAX_MSG_RATE_PERCORE)));
        return flinkConfig;
    }
writeConfigToFile

@param configJobDirectory job config directory
@param configFileName config file name
@param content contents of the file to be written
@return whether success
inlong-manager/manager-plugins/base/src/main/java/org/apache/inlong/manager/plugin/util/FlinkUtils.java