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.StandaloneClusterId;
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.client.program.rest.RestClusterClient;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.springblade.etl.engine.flink.dto.PackageJarJob;
import org.springblade.etl.engine.flink.util.FlinkKit;

public class RestClusterClientCommand implements SubmitCommand {

    @Override
    public Tuple2<String, ClusterClient> submit(Path flinkHome, Configuration configuration, PackageJarJob job) throws Exception {
        RestClusterClient<StandaloneClusterId> client = createClusterClient(configuration);
        PackagedProgram program = FlinkKit.buildProgram(configuration, job);
        JobGraph jobGraph = PackagedProgramUtils.createJobGraph(program, configuration, job.getParallelism(), false);
        JobID jobID = client.submitJob(jobGraph).get();
        return new Tuple2<>(jobID.toHexString(), client);
    }

    private RestClusterClient<StandaloneClusterId> createClusterClient(Configuration configuration) throws Exception {
//    	configuration.setLong(RestOptions.CONNECTION_TIMEOUT, 10_000);
//    	configuration.setLong(RestOptions.AWAIT_LEADER_TIMEOUT, 10_000);
//    	configuration.setLong(RestOptions.RETRY_DELAY, 0);
//    	configuration.setInteger(RestOptions.RETRY_MAX_ATTEMPTS, 0);
        return new RestClusterClient<>(configuration, StandaloneClusterId.getInstance());
    }
}
