[Bug][Manager] resource may leak due to BufferedReader not closed

[aloyszhang]

What happened

https://github.com/apache/inlong/blob/master/inlong-manager/manager-plugins/base/src/main/java/org/apache/inlong/manager/plugin/util/FlinkConfiguration.java#L98~L99

private FlinkConfig getFlinkConfigFromFile(String fileName) throws IOException {
        Properties properties = new Properties();
        BufferedReader bufferedReader = new BufferedReader(new FileReader(fileName));
        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.setAuditProxyHosts(properties.getProperty(METRICS_AUDIT_PROXY_HOSTS_KEY));
        flinkConfig.setVersion(properties.getProperty(FLINK_VERSION));
        return flinkConfig;
    }

Some places are using a BufferedReader and do not close properly.

What you expected to happen

BufferedReader should be closed after use

How to reproduce

See:

https://github.com/apache/inlong/blob/master/inlong-manager/manager-plugins/base/src/main/java/org/apache/inlong/manager/plugin/util/FlinkConfiguration.java#L98~L99

private FlinkConfig getFlinkConfigFromFile(String fileName) throws IOException {
        Properties properties = new Properties();
        BufferedReader bufferedReader = new BufferedReader(new FileReader(fileName));
        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.setAuditProxyHosts(properties.getProperty(METRICS_AUDIT_PROXY_HOSTS_KEY));
        flinkConfig.setVersion(properties.getProperty(FLINK_VERSION));
        return flinkConfig;
    }

Environment

No response

InLong version

master

InLong Component

InLong Manager

Are you willing to submit PR?

• Yes, I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct