flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/JobLeaderService.java
/**
	 * Add the given job to be monitored. This means that the service tries to detect leaders for
	 * this job and then tries to establish a connection to it.
	 *
	 * @param jobId identifying the job to monitor
	 * @param defaultTargetAddress of the job leader
	 * @throws Exception if an error occurs while starting the leader retrieval service
	 */
Add the given job to be monitored. This means that the service tries to detect leaders for
this job and then tries to establish a connection to it.

@param jobId identifying the job to monitor
@param defaultTargetAddress of the job leader
@throws Exception if an error occurs while starting the leader retrieval service
Add the given job to be monitored . This means that the service tries to detect leaders for this job and then tries to establish a connection to it . @param jobId identifying the job to monitor @param defaultTargetAddress of the job leader @throws Exception if an error occurs while starting the leader retrieval service
public void addJob(final JobID jobId, final String defaultTargetAddress) throws Exception {
		Preconditions.checkState(JobLeaderService.State.STARTED == state, "The service is currently not running.");
		LOG.info("Add job {} for job leader monitoring.", jobId);
		final LeaderRetrievalService leaderRetrievalService = highAvailabilityServices.getJobManagerLeaderRetriever(
			jobId,
			defaultTargetAddress);
		JobLeaderService.JobManagerLeaderListener jobManagerLeaderListener = new JobManagerLeaderListener(jobId);
		leaderRetrievalService.start(jobManagerLeaderListener);
		jobLeaderServices.put(jobId, Tuple2.of(leaderRetrievalService, jobManagerLeaderListener));
	}
public void addJob ( final JobID jobId , final String defaultTargetAddress ) throws Exception { Preconditions . checkState ( JobLeaderService . State . STARTED == state , "The service is currently not running." ) ; LOG . info ( "Add job {} for job leader monitoring." , jobId ) ; final LeaderRetrievalService leaderRetrievalService = highAvailabilityServices . getJobManagerLeaderRetriever ( jobId , defaultTargetAddress ) ; JobLeaderService . JobManagerLeaderListener jobManagerLeaderListener = new JobManagerLeaderListener ( jobId ) ; leaderRetrievalService . start ( jobManagerLeaderListener ) ; jobLeaderServices . put ( jobId , Tuple2 . of ( leaderRetrievalService , jobManagerLeaderListener ) ) ; }
Add the given job to be monitored.
Add the given job to be monitored .