//package org.springblade.etl.engine.flink.client.submit;
//
//import java.nio.file.Path;
//
//import org.apache.flink.api.common.JobID;
//import org.apache.flink.api.java.tuple.Tuple2;
//import org.apache.flink.client.deployment.ClusterClientFactory;
//import org.apache.flink.client.program.ClusterClient;
//import org.apache.flink.client.program.PackagedProgram;
//import org.apache.flink.client.program.PackagedProgramUtils;
//import org.apache.flink.configuration.Configuration;
//import org.apache.flink.runtime.jobgraph.JobGraph;
//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 lombok.extern.slf4j.Slf4j;
//
//@Slf4j
//public class YarnSessionSubmitCommand implements SubmitCommand {
//
//    @Override
//    public Tuple2<String, ClusterClient> submit(Path flinkHome, Configuration configuration, PackageJarJob job) throws Exception {
//        ClusterClientFactory<ApplicationId> factory = FlinkKit.createClientFactory(configuration);
//        YarnClusterDescriptor clusterDescriptor = (YarnClusterDescriptor) FlinkKit.createClusterDescriptor(factory, configuration);
//
//        ApplicationId clusterId = factory.getClusterId(configuration);
//        ClusterClient<ApplicationId> clusterClient = clusterDescriptor.retrieve(clusterId).getClusterClient();
//
//        PackagedProgram program = FlinkKit.buildProgram(configuration, job);
//        JobGraph jobGraph = PackagedProgramUtils.createJobGraph(program, configuration, 1, false);
//        JobID jobID = clusterClient.submitJob(jobGraph).get();
//        return new Tuple2<>(jobID.toHexString(), clusterClient);
//    }
//
//}
