flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobClient.java
/**
	 * Submits a job in detached mode. The method sends the JobGraph to the
	 * JobManager and waits for the answer whether the job could be started or not.
	 *
	 * @param jobManagerGateway Gateway to the JobManager which will execute the jobs
	 * @param config The cluster wide configuration.
	 * @param jobGraph The job
	 * @param timeout Timeout in which the JobManager must have responded.
	 */
Submits a job in detached mode. The method sends the JobGraph to the
JobManager and waits for the answer whether the job could be started or not.

@param jobManagerGateway Gateway to the JobManager which will execute the jobs
@param config The cluster wide configuration.
@param jobGraph The job
@param timeout Timeout in which the JobManager must have responded.
Submits a job in detached mode . The method sends the JobGraph to the JobManager and waits for the answer whether the job could be started or not . @param jobManagerGateway Gateway to the JobManager which will execute the jobs @param config The cluster wide configuration . @param jobGraph The job @param timeout Timeout in which the JobManager must have responded .
public static void submitJobDetached(
			JobManagerGateway jobManagerGateway,
			Configuration config,
			JobGraph jobGraph,
			Time timeout,
			ClassLoader classLoader) throws JobExecutionException {
		checkNotNull(jobManagerGateway, "The jobManagerGateway must not be null.");
		checkNotNull(jobGraph, "The jobGraph must not be null.");
		checkNotNull(timeout, "The timeout must not be null.");
		LOG.info("Checking and uploading JAR files");
		final CompletableFuture<InetSocketAddress> blobServerAddressFuture = retrieveBlobServerAddress(
			jobManagerGateway,
			timeout);
		final InetSocketAddress blobServerAddress;
		try {
			blobServerAddress = blobServerAddressFuture.get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
		} catch (Exception e) {
			throw new JobSubmissionException(jobGraph.getJobID(), "Could not retrieve BlobServer address.", e);
		}
		try {
			jobGraph.uploadUserJars(blobServerAddress, config);
		}
		catch (IOException e) {
			throw new JobSubmissionException(jobGraph.getJobID(),
				"Could not upload the program's JAR files to the JobManager.", e);
		}
		CompletableFuture<Acknowledge> submissionFuture = jobManagerGateway.submitJob(jobGraph, ListeningBehaviour.DETACHED, timeout);
		try {
			submissionFuture.get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
		} catch (TimeoutException e) {
			throw new JobTimeoutException(jobGraph.getJobID(),
				"JobManager did not respond within " + timeout, e);
		} catch (Throwable throwable) {
			Throwable stripped = ExceptionUtils.stripExecutionException(throwable);
			try {
				ExceptionUtils.tryDeserializeAndThrow(stripped, classLoader);
			} catch (JobExecutionException jee) {
				throw jee;
			} catch (Throwable t) {
				throw new JobExecutionException(
					jobGraph.getJobID(),
					"JobSubmission failed.",
					t);
			}
		}
	}
public static void submitJobDetached ( JobManagerGateway jobManagerGateway , Configuration config , JobGraph jobGraph , Time timeout , ClassLoader classLoader ) throws JobExecutionException { checkNotNull ( jobManagerGateway , "The jobManagerGateway must not be null." ) ; checkNotNull ( jobGraph , "The jobGraph must not be null." ) ; checkNotNull ( timeout , "The timeout must not be null." ) ; LOG . info ( "Checking and uploading JAR files" ) ; final CompletableFuture < InetSocketAddress > blobServerAddressFuture = retrieveBlobServerAddress ( jobManagerGateway , timeout ) ; final InetSocketAddress blobServerAddress ; try { blobServerAddress = blobServerAddressFuture . get ( timeout . toMilliseconds ( ) , TimeUnit . MILLISECONDS ) ; } catch ( Exception e ) { throw new JobSubmissionException ( jobGraph . getJobID ( ) , "Could not retrieve BlobServer address." , e ) ; } try { jobGraph . uploadUserJars ( blobServerAddress , config ) ; } catch ( IOException e ) { throw new JobSubmissionException ( jobGraph . getJobID ( ) , "Could not upload the program's JAR files to the JobManager." , e ) ; } CompletableFuture < Acknowledge > submissionFuture = jobManagerGateway . submitJob ( jobGraph , ListeningBehaviour . DETACHED , timeout ) ; try { submissionFuture . get ( timeout . toMilliseconds ( ) , TimeUnit . MILLISECONDS ) ; } catch ( TimeoutException e ) { throw new JobTimeoutException ( jobGraph . getJobID ( ) , "JobManager did not respond within " + timeout , e ) ; } catch ( Throwable throwable ) { Throwable stripped = ExceptionUtils . stripExecutionException ( throwable ) ; try { ExceptionUtils . tryDeserializeAndThrow ( stripped , classLoader ) ; } catch ( JobExecutionException jee ) { throw jee ; } catch ( Throwable t ) { throw new JobExecutionException ( jobGraph . getJobID ( ) , "JobSubmission failed." , t ) ; } } }
Submits a job in detached mode.
Submits a job in detached mode .