flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobClient.java
/**
	 * Attaches to a running Job using the JobID.
	 * Reconstructs the user class loader by downloading the jars from the JobManager.
	 */
Attaches to a running Job using the JobID.
Reconstructs the user class loader by downloading the jars from the JobManager.
Attaches to a running Job using the JobID . Reconstructs the user class loader by downloading the jars from the JobManager .
public static JobListeningContext attachToRunningJob(
			JobID jobID,
			Configuration configuration,
			ActorSystem actorSystem,
			HighAvailabilityServices highAvailabilityServices,
			FiniteDuration timeout,
			boolean sysoutLogUpdates) {
		checkNotNull(jobID, "The jobID must not be null.");
		checkNotNull(configuration, "The configuration must not be null.");
		checkNotNull(actorSystem, "The actorSystem must not be null.");
		checkNotNull(highAvailabilityServices, "The high availability services must not be null.");
		checkNotNull(timeout, "The timeout must not be null.");
		Props jobClientActorProps = JobAttachmentClientActor.createActorProps(
			highAvailabilityServices.getJobManagerLeaderRetriever(HighAvailabilityServices.DEFAULT_JOB_ID),
			timeout,
			sysoutLogUpdates);
		ActorRef jobClientActor = actorSystem.actorOf(jobClientActorProps);
		Future<Object> attachmentFuture = Patterns.ask(
				jobClientActor,
				new JobClientMessages.AttachToJobAndWait(jobID),
				new Timeout(AkkaUtils.INF_TIMEOUT()));
		return new JobListeningContext(
			jobID,
			attachmentFuture,
			jobClientActor,
			timeout,
			actorSystem,
			configuration,
			highAvailabilityServices);
	}
public static JobListeningContext attachToRunningJob ( JobID jobID , Configuration configuration , ActorSystem actorSystem , HighAvailabilityServices highAvailabilityServices , FiniteDuration timeout , boolean sysoutLogUpdates ) { checkNotNull ( jobID , "The jobID must not be null." ) ; checkNotNull ( configuration , "The configuration must not be null." ) ; checkNotNull ( actorSystem , "The actorSystem must not be null." ) ; checkNotNull ( highAvailabilityServices , "The high availability services must not be null." ) ; checkNotNull ( timeout , "The timeout must not be null." ) ; Props jobClientActorProps = JobAttachmentClientActor . createActorProps ( highAvailabilityServices . getJobManagerLeaderRetriever ( HighAvailabilityServices . DEFAULT_JOB_ID ) , timeout , sysoutLogUpdates ) ; ActorRef jobClientActor = actorSystem . actorOf ( jobClientActorProps ) ; Future < Object > attachmentFuture = Patterns . ask ( jobClientActor , new JobClientMessages . AttachToJobAndWait ( jobID ) , new Timeout ( AkkaUtils . INF_TIMEOUT ( ) ) ) ; return new JobListeningContext ( jobID , attachmentFuture , jobClientActor , timeout , actorSystem , configuration , highAvailabilityServices ) ; }
Attaches to a running Job using the JobID.
Attaches to a running Job using the JobID .
// we create a proxy JobClientActor that deals with all communication with // the JobManager. It forwards the job attachments, checks the success/failure responses, logs // update messages, watches for disconnect between client and JobManager, ...