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 to a Flink cluster (non-blocking) and returns a JobListeningContext which can be
	 * passed to {@code awaitJobResult} to get the result of the submission.
	 * @return JobListeningContext which may be used to retrieve the JobExecutionResult via
	 * 			{@code awaitJobResult(JobListeningContext context)}.
	 */
Submits a job to a Flink cluster (non-blocking) and returns a JobListeningContext which can be
passed to  awaitJobResult to get the result of the submission.
Submits a job to a Flink cluster ( non - blocking ) and returns a JobListeningContext which can be passed to awaitJobResult to get the result of the submission .
public static JobListeningContext submitJob(
			ActorSystem actorSystem,
			Configuration config,
			HighAvailabilityServices highAvailabilityServices,
			JobGraph jobGraph,
			FiniteDuration timeout,
			boolean sysoutLogUpdates,
			ClassLoader classLoader) {
		checkNotNull(actorSystem, "The actorSystem must not be null.");
		checkNotNull(highAvailabilityServices, "The high availability services must not be null.");
		checkNotNull(jobGraph, "The jobGraph must not be null.");
		checkNotNull(timeout, "The timeout must not be null.");
		Props jobClientActorProps = JobSubmissionClientActor.createActorProps(
			highAvailabilityServices.getJobManagerLeaderRetriever(HighAvailabilityServices.DEFAULT_JOB_ID),
			timeout,
			sysoutLogUpdates,
			config);
		ActorRef jobClientActor = actorSystem.actorOf(jobClientActorProps);
		Future<Object> submissionFuture = Patterns.ask(
				jobClientActor,
				new JobClientMessages.SubmitJobAndWait(jobGraph),
				new Timeout(AkkaUtils.INF_TIMEOUT()));
		return new JobListeningContext(
			jobGraph.getJobID(),
			submissionFuture,
			jobClientActor,
			timeout,
			classLoader,
			highAvailabilityServices);
	}
public static JobListeningContext submitJob ( ActorSystem actorSystem , Configuration config , HighAvailabilityServices highAvailabilityServices , JobGraph jobGraph , FiniteDuration timeout , boolean sysoutLogUpdates , ClassLoader classLoader ) { checkNotNull ( actorSystem , "The actorSystem must not be null." ) ; checkNotNull ( highAvailabilityServices , "The high availability services must not be null." ) ; checkNotNull ( jobGraph , "The jobGraph must not be null." ) ; checkNotNull ( timeout , "The timeout must not be null." ) ; Props jobClientActorProps = JobSubmissionClientActor . createActorProps ( highAvailabilityServices . getJobManagerLeaderRetriever ( HighAvailabilityServices . DEFAULT_JOB_ID ) , timeout , sysoutLogUpdates , config ) ; ActorRef jobClientActor = actorSystem . actorOf ( jobClientActorProps ) ; Future < Object > submissionFuture = Patterns . ask ( jobClientActor , new JobClientMessages . SubmitJobAndWait ( jobGraph ) , new Timeout ( AkkaUtils . INF_TIMEOUT ( ) ) ) ; return new JobListeningContext ( jobGraph . getJobID ( ) , submissionFuture , jobClientActor , timeout , classLoader , highAvailabilityServices ) ; }
Submits a job to a Flink cluster (non-blocking) and returns a JobListeningContext which can be
 passed to {@code awaitJobResult} to get the result of the submission.
Submits a job to a Flink cluster ( non - blocking ) and returns a JobListeningContext which can be passed to { @code awaitJobResult } to get the result of the submission .
// for this job, we create a proxy JobClientActor that deals with all communication with // the JobManager. It forwards the job submission, checks the success/failure responses, logs // update messages, watches for disconnect between client and JobManager, ...