flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/client/JobClientActorTest.java
/** Tests that JobClient throws an Exception if the JobClientActor dies and can't answer to
	 * {@link akka.actor.Identify} message anymore.
	 */
Tests that JobClient throws an Exception if the JobClientActor dies and can't answer to
akka.actor.Identify message anymore.
Tests that JobClient throws an Exception if the JobClientActor dies and can ' t answer to akka . actor . Identify message anymore .
@Test
	public void testGuaranteedAnswerIfJobClientDies() throws Exception {
		FiniteDuration timeout = new FiniteDuration(2L, TimeUnit.SECONDS);
			UUID leaderSessionID = UUID.randomUUID();
		ActorRef jobManager = system.actorOf(
			Props.create(
				JobAcceptingActor.class,
				leaderSessionID));
		TestingLeaderRetrievalService testingLeaderRetrievalService = new TestingLeaderRetrievalService(
			jobManager.path().toString(),
			leaderSessionID
		);
		TestingHighAvailabilityServices highAvailabilityServices = new TestingHighAvailabilityServices();
		highAvailabilityServices.setJobMasterLeaderRetriever(HighAvailabilityServices.DEFAULT_JOB_ID, testingLeaderRetrievalService);
		JobListeningContext jobListeningContext =
			JobClient.submitJob(
				system,
				clientConfig,
				highAvailabilityServices,
				testJobGraph,
				timeout,
				false,
				getClass().getClassLoader());
		Future<Object> waitFuture = Patterns.ask(jobManager, new RegisterTest(), new Timeout(timeout));
		Await.result(waitFuture, timeout);
		jobListeningContext.getJobClientActor().tell(PoisonPill.getInstance(), ActorRef.noSender());
		try {
			JobClient.awaitJobResult(jobListeningContext);
			Assert.fail();
		} catch (JobExecutionException e) {
		} finally {
			highAvailabilityServices.closeAndCleanupAllData();
		}
	}
@ Test public void testGuaranteedAnswerIfJobClientDies ( ) throws Exception { FiniteDuration timeout = new FiniteDuration ( 2L , TimeUnit . SECONDS ) ; UUID leaderSessionID = UUID . randomUUID ( ) ; ActorRef jobManager = system . actorOf ( Props . create ( JobAcceptingActor . class , leaderSessionID ) ) ; TestingLeaderRetrievalService testingLeaderRetrievalService = new TestingLeaderRetrievalService ( jobManager . path ( ) . toString ( ) , leaderSessionID ) ; TestingHighAvailabilityServices highAvailabilityServices = new TestingHighAvailabilityServices ( ) ; highAvailabilityServices . setJobMasterLeaderRetriever ( HighAvailabilityServices . DEFAULT_JOB_ID , testingLeaderRetrievalService ) ; JobListeningContext jobListeningContext = JobClient . submitJob ( system , clientConfig , highAvailabilityServices , testJobGraph , timeout , false , getClass ( ) . getClassLoader ( ) ) ; Future < Object > waitFuture = Patterns . ask ( jobManager , new RegisterTest ( ) , new Timeout ( timeout ) ) ; Await . result ( waitFuture , timeout ) ; jobListeningContext . getJobClientActor ( ) . tell ( PoisonPill . getInstance ( ) , ActorRef . noSender ( ) ) ; try { JobClient . awaitJobResult ( jobListeningContext ) ; Assert . fail ( ) ; } catch ( JobExecutionException e ) { } finally { highAvailabilityServices . closeAndCleanupAllData ( ) ; } }
Tests that JobClient throws an Exception if the JobClientActor dies and can't answer to
 {@link akka.actor.Identify} message anymore.
Tests that JobClient throws an Exception if the JobClientActor dies and can ' t answer to { @link akka . actor . Identify } message anymore .
// kill the job client actor which has been registered at the JobManager // should not block but return an error // this is what we want