public static Object getFlinkClientService(Configuration configuration, FlinkConfig flinkConfig) {
        log.info("Flink version {}", flinkConfig.getVersion());

        Path pluginPath = Paths.get(DEFAULT_PLUGINS).toAbsolutePath();
        String flinkJarName = String.format(Constants.FLINK_JAR_NAME, flinkConfig.getVersion());
        String flinkClientPath = FILE_PREFIX + pluginPath + File.separator + flinkJarName;
        log.info("Start to load Flink jar: {}", flinkClientPath);

        try (URLClassLoader classLoader = new URLClassLoader(new URL[]{new URL(flinkClientPath)}, Thread.currentThread()
                .getContextClassLoader())) {
            Class<?> flinkClientService = classLoader.loadClass(Constants.FLINK_CLIENT_CLASS);
            Object flinkService = flinkClientService.getDeclaredConstructor(Configuration.class)
                    .newInstance(configuration);
            log.info("Successfully loaded Flink service");
            return flinkService;
        } catch (Exception e) {
            log.error("Failed to loaded Flink service, please check flink client jar path: {}", flinkClientPath);
            throw new RuntimeException(e);
        }
    }
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