JobClientActor extends FlinkUntypedActor implements LeaderRetrievalListener { public static Props createJobClientActorProps( LeaderRetrievalService leaderRetrievalService, FiniteDuration timeout, boolean sysoutUpdates) { return Props.create( JobClientActor.class, leaderRetrievalService, timeout, sysoutUpdates); } JobClientActor(
			LeaderRetrievalService leaderRetrievalService,
			FiniteDuration timeout,
			boolean sysoutUpdates); @Override void preStart(); @Override void postStop(); @Override void notifyLeaderAddress(String leaderAddress, UUID leaderSessionID); @Override void handleError(Exception exception); static Props createJobClientActorProps(
			LeaderRetrievalService leaderRetrievalService,
			FiniteDuration timeout,
			boolean sysoutUpdates);  }
@Test(expected=JobClientActorSubmissionTimeoutException.class) public void testSubmissionTimeout() throws Exception { FiniteDuration jobClientActorTimeout = new FiniteDuration(5, TimeUnit.SECONDS); FiniteDuration timeout = jobClientActorTimeout.$times(2); UUID leaderSessionID = UUID.randomUUID(); ActorRef jobManager = system.actorOf( Props.create( PlainActor.class, leaderSessionID)); TestingLeaderRetrievalService testingLeaderRetrievalService = new TestingLeaderRetrievalService( jobManager.path().toString(), leaderSessionID ); Props jobClientActorProps = JobClientActor.createJobClientActorProps( testingLeaderRetrievalService, jobClientActorTimeout, false); ActorRef jobClientActor = system.actorOf(jobClientActorProps); Future<Object> jobExecutionResult = Patterns.ask( jobClientActor, new JobClientMessages.SubmitJobAndWait(testJobGraph), new Timeout(timeout)); Await.result(jobExecutionResult, timeout); }
@Test(expected=JobClientActorConnectionTimeoutException.class) public void testConnectionTimeoutWithoutJobManager() throws Exception { FiniteDuration jobClientActorTimeout = new FiniteDuration(5, TimeUnit.SECONDS); FiniteDuration timeout = jobClientActorTimeout.$times(2); TestingLeaderRetrievalService testingLeaderRetrievalService = new TestingLeaderRetrievalService(); Props jobClientActorProps = JobClientActor.createJobClientActorProps( testingLeaderRetrievalService, jobClientActorTimeout, false); ActorRef jobClientActor = system.actorOf(jobClientActorProps); Future<Object> jobExecutionResult = Patterns.ask( jobClientActor, new JobClientMessages.SubmitJobAndWait(testJobGraph), new Timeout(timeout)); Await.result(jobExecutionResult, timeout); }
@Test(expected=JobClientActorConnectionTimeoutException.class) public void testConnectionTimeoutAfterJobSubmission() throws Exception { FiniteDuration jobClientActorTimeout = new FiniteDuration(5, TimeUnit.SECONDS); FiniteDuration timeout = jobClientActorTimeout.$times(2); UUID leaderSessionID = UUID.randomUUID(); ActorRef jobManager = system.actorOf( Props.create( JobAcceptingActor.class, leaderSessionID)); TestingLeaderRetrievalService testingLeaderRetrievalService = new TestingLeaderRetrievalService( jobManager.path().toString(), leaderSessionID ); Props jobClientActorProps = JobClientActor.createJobClientActorProps( testingLeaderRetrievalService, jobClientActorTimeout, false); ActorRef jobClientActor = system.actorOf(jobClientActorProps); Future<Object> jobExecutionResult = Patterns.ask( jobClientActor, new JobClientMessages.SubmitJobAndWait(testJobGraph), new Timeout(timeout)); Future<Object> waitFuture = Patterns.ask(jobManager, new RegisterTest(), new Timeout(timeout)); Await.result(waitFuture, timeout); jobManager.tell(PoisonPill.getInstance(), ActorRef.noSender()); Await.result(jobExecutionResult, timeout); }