flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobClient.java
/**
	 * Given a JobListeningContext, awaits the result of the job execution that this context is bound to
	 * @param listeningContext The listening context of the job execution
	 * @return The result of the execution
	 * @throws JobExecutionException if anything goes wrong while monitoring the job
	 */
Given a JobListeningContext, awaits the result of the job execution that this context is bound to
@param listeningContext The listening context of the job execution
@return The result of the execution
@throws JobExecutionException if anything goes wrong while monitoring the job
Given a JobListeningContext awaits the result of the job execution that this context is bound to @param listeningContext The listening context of the job execution @return The result of the execution @throws JobExecutionException if anything goes wrong while monitoring the job
public static JobExecutionResult awaitJobResult(JobListeningContext listeningContext) throws JobExecutionException {
		final JobID jobID = listeningContext.getJobID();
		final ActorRef jobClientActor = listeningContext.getJobClientActor();
		final Future<Object> jobSubmissionFuture = listeningContext.getJobResultFuture();
		final FiniteDuration askTimeout = listeningContext.getTimeout();
		final ClassLoader classLoader = listeningContext.getClassLoader();
		while (!jobSubmissionFuture.isCompleted()) {
			try {
				Await.ready(jobSubmissionFuture, askTimeout);
			} catch (InterruptedException e) {
				throw new JobExecutionException(
					jobID,
					"Interrupted while waiting for job completion.");
			} catch (TimeoutException e) {
				try {
					Await.result(
						Patterns.ask(
							jobClientActor,
							new Identify(true),
							Timeout.durationToTimeout(askTimeout)),
						askTimeout);
				} catch (Exception eInner) {
					if (!jobSubmissionFuture.isCompleted()) {
						throw new JobExecutionException(
							jobID,
							"JobClientActor seems to have died before the JobExecutionResult could be retrieved.",
							eInner);
					}
				}
			}
		}
		final Object answer;
		try {
			answer = Await.result(jobSubmissionFuture, Duration.Zero());
		}
		catch (Throwable throwable) {
			throw new JobExecutionException(jobID,
				"Couldn't retrieve the JobExecutionResult from the JobManager.", throwable);
		}
		finally {
			jobClientActor.tell(PoisonPill.getInstance(), ActorRef.noSender());
		}
		if (answer instanceof JobManagerMessages.JobResultSuccess) {
			LOG.info("Job execution complete");
			SerializedJobExecutionResult result = ((JobManagerMessages.JobResultSuccess) answer).result();
			if (result != null) {
				try {
					return result.toJobExecutionResult(classLoader);
				} catch (Throwable t) {
					throw new JobExecutionException(jobID,
						"Job was successfully executed but JobExecutionResult could not be deserialized.");
				}
			} else {
				throw new JobExecutionException(jobID,
					"Job was successfully executed but result contained a null JobExecutionResult.");
			}
		}
		else if (answer instanceof JobManagerMessages.JobResultFailure) {
			LOG.info("Job execution failed");
			SerializedThrowable serThrowable = ((JobManagerMessages.JobResultFailure) answer).cause();
			if (serThrowable != null) {
				Throwable cause = serThrowable.deserializeError(classLoader);
				if (cause instanceof JobExecutionException) {
					throw (JobExecutionException) cause;
				} else {
					throw new JobExecutionException(jobID, "Job execution failed", cause);
				}
			} else {
				throw new JobExecutionException(jobID,
					"Job execution failed with null as failure cause.");
			}
		}
		else if (answer instanceof JobManagerMessages.JobNotFound) {
			throw new JobRetrievalException(
				((JobManagerMessages.JobNotFound) answer).jobID(),
				"Couldn't retrieve Job " + jobID + " because it was not running.");
		}
		else {
			throw new JobExecutionException(jobID,
				"Unknown answer from JobManager after submitting the job: " + answer);
		}
	}
public static JobExecutionResult awaitJobResult ( JobListeningContext listeningContext ) throws JobExecutionException { final JobID jobID = listeningContext . getJobID ( ) ; final ActorRef jobClientActor = listeningContext . getJobClientActor ( ) ; final Future < Object > jobSubmissionFuture = listeningContext . getJobResultFuture ( ) ; final FiniteDuration askTimeout = listeningContext . getTimeout ( ) ; final ClassLoader classLoader = listeningContext . getClassLoader ( ) ; while ( ! jobSubmissionFuture . isCompleted ( ) ) { try { Await . ready ( jobSubmissionFuture , askTimeout ) ; } catch ( InterruptedException e ) { throw new JobExecutionException ( jobID , "Interrupted while waiting for job completion." ) ; } catch ( TimeoutException e ) { try { Await . result ( Patterns . ask ( jobClientActor , new Identify ( true ) , Timeout . durationToTimeout ( askTimeout ) ) , askTimeout ) ; } catch ( Exception eInner ) { if ( ! jobSubmissionFuture . isCompleted ( ) ) { throw new JobExecutionException ( jobID , "JobClientActor seems to have died before the JobExecutionResult could be retrieved." , eInner ) ; } } } } final Object answer ; try { answer = Await . result ( jobSubmissionFuture , Duration . Zero ( ) ) ; } catch ( Throwable throwable ) { throw new JobExecutionException ( jobID , "Couldn't retrieve the JobExecutionResult from the JobManager." , throwable ) ; } finally { jobClientActor . tell ( PoisonPill . getInstance ( ) , ActorRef . noSender ( ) ) ; } if ( answer instanceof JobManagerMessages . JobResultSuccess ) { LOG . info ( "Job execution complete" ) ; SerializedJobExecutionResult result = ( ( JobManagerMessages . JobResultSuccess ) answer ) . result ( ) ; if ( result != null ) { try { return result . toJobExecutionResult ( classLoader ) ; } catch ( Throwable t ) { throw new JobExecutionException ( jobID , "Job was successfully executed but JobExecutionResult could not be deserialized." ) ; } } else { throw new JobExecutionException ( jobID , "Job was successfully executed but result contained a null JobExecutionResult." ) ; } } else if ( answer instanceof JobManagerMessages . JobResultFailure ) { LOG . info ( "Job execution failed" ) ; SerializedThrowable serThrowable = ( ( JobManagerMessages . JobResultFailure ) answer ) . cause ( ) ; if ( serThrowable != null ) { Throwable cause = serThrowable . deserializeError ( classLoader ) ; if ( cause instanceof JobExecutionException ) { throw ( JobExecutionException ) cause ; } else { throw new JobExecutionException ( jobID , "Job execution failed" , cause ) ; } } else { throw new JobExecutionException ( jobID , "Job execution failed with null as failure cause." ) ; } } else if ( answer instanceof JobManagerMessages . JobNotFound ) { throw new JobRetrievalException ( ( ( JobManagerMessages . JobNotFound ) answer ) . jobID ( ) , "Couldn't retrieve Job " + jobID + " because it was not running." ) ; } else { throw new JobExecutionException ( jobID , "Unknown answer from JobManager after submitting the job: " + answer ) ; } }
Given a JobListeningContext, awaits the result of the job execution that this context is bound to
 @param listeningContext The listening context of the job execution
 @return The result of the execution
 @throws JobExecutionException if anything goes wrong while monitoring the job
Given a JobListeningContext awaits the result of the job execution that this context is bound to @param listeningContext The listening context of the job execution @return The result of the execution @throws JobExecutionException if anything goes wrong while monitoring the job
// retrieves class loader if necessary // wait for the future which holds the result to be ready // ping the JobClientActor from time to time to check if it is still running // Ping the Actor to see if it is alive // we got a reply, continue waiting for the job result // we could have a result but the JobClientActor might have been killed and // thus the health check failed // we have already awaited the result, zero time to wait here // failsafe shutdown of the client actor // second block handles the actual response