//package org.springblade.etl.engine.flink.client.submit;
//
//import java.nio.file.Path;
//
//import org.apache.flink.api.java.tuple.Tuple2;
//import org.apache.flink.client.deployment.ClusterDeploymentException;
//import org.apache.flink.client.deployment.ClusterSpecification;
//import org.apache.flink.client.deployment.application.ApplicationConfiguration;
//import org.apache.flink.client.program.ClusterClient;
//import org.apache.flink.client.program.ClusterClientProvider;
//import org.apache.flink.configuration.Configuration;
//import org.apache.flink.yarn.YarnClusterDescriptor;
//import org.apache.hadoop.yarn.api.records.ApplicationId;
//import org.springblade.etl.engine.flink.dto.PackageJarJob;
//import org.springblade.etl.engine.flink.util.FlinkKit;
//import org.springblade.etl.engine.flink.util.Util;
//
//import lombok.extern.slf4j.Slf4j;
//
//@Slf4j
//public class YarnApplicationCommand implements SubmitCommand {
//
//    @Override
//    public Tuple2<String, ClusterClient> submit(Path flinkHome, Configuration configuration, PackageJarJob job) throws Exception {
//        YarnClusterDescriptor clusterDescriptor = (YarnClusterDescriptor) FlinkKit.createClusterDescriptor(configuration);
//        Util.addJarFiles(clusterDescriptor, flinkHome, configuration);
//        ClusterSpecification clusterSpecification = FlinkKit.createClusterSpecification(configuration);
//
//        ApplicationConfiguration applicationConfiguration = new ApplicationConfiguration(job.getProgramArgs(), job.getEntryPointClass());
//        ClusterClient<ApplicationId> clusterClient = createClusterClient(clusterDescriptor, clusterSpecification, applicationConfiguration);
//        return new Tuple2<>(clusterClient.getClusterId().toString(), clusterClient);
//    }
//
//    private ClusterClient<ApplicationId> createClusterClient(YarnClusterDescriptor clusterDescriptor,
//                                                             ClusterSpecification clusterSpecification,
//                                                             ApplicationConfiguration applicationConfiguration) throws ClusterDeploymentException {
//        ClusterClientProvider<ApplicationId> provider = clusterDescriptor.deployApplicationCluster(clusterSpecification, applicationConfiguration);
//        ClusterClient<ApplicationId> clusterClient = provider.getClusterClient();
//        log.info("deploy application with appId: {}", clusterClient.getClusterId());
//        return clusterClient;
//    }
//    
//}
