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 ) ; final Tuple2 < LeaderRetrievalService , JobManagerLeaderListener > oldEntry = jobLeaderServices . put ( jobId , Tuple2 . of ( leaderRetrievalService , jobManagerLeaderListener ) ) ; if ( oldEntry != null ) { oldEntry . f0 . stop ( ) ; oldEntry . f1 . stop ( ) ; } leaderRetrievalService . start ( jobManagerLeaderListener ) ; }
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 .