tillrohrmann closed pull request #6811: [BP-1.6][FLINK-9788] Fix ExecutionGraph inconsistency for global failures when restarting URL: https://github.com/apache/flink/pull/6811
This is a PR merged from a forked repository. As GitHub hides the original diff on merge, it is displayed below for the sake of provenance: As this is a foreign pull request (from a fork), the diff is supplied below (as it won't show otherwise due to GitHub magic): diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java index acb1e16fe71..0be1ff27420 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java @@ -1151,18 +1151,7 @@ public void failGlobal(Throwable t) { current == JobStatus.SUSPENDED || current.isGloballyTerminalState()) { return; - } - else if (current == JobStatus.RESTARTING) { - // we handle 'failGlobal()' while in 'RESTARTING' as a safety net in case something - // has gone wrong in 'RESTARTING' and we need to re-attempt the restarts - initFailureCause(t); - - final long globalVersionForRestart = incrementGlobalModVersion(); - if (tryRestartOrFail(globalVersionForRestart)) { - return; - } - } - else if (transitionState(current, JobStatus.FAILING, t)) { + } else if (transitionState(current, JobStatus.FAILING, t)) { initFailureCause(t); // make sure no concurrent local or global actions interfere with the failover @@ -1240,7 +1229,7 @@ public void restart(long expectedGlobalVersion) { colGroups.add(cgroup); } - jv.resetForNewExecution(resetTimestamp, globalModVersion); + jv.resetForNewExecution(resetTimestamp, expectedGlobalVersion); } for (int i = 0; i < stateTimestamps.length; i++) { diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphRestartTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphRestartTest.java index 9b98de78143..91510d1af54 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphRestartTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphRestartTest.java @@ -25,6 +25,7 @@ import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.configuration.Configuration; import org.apache.flink.core.testutils.OneShotLatch; +import org.apache.flink.runtime.JobException; import org.apache.flink.runtime.akka.AkkaUtils; import org.apache.flink.runtime.clusterframework.types.ResourceID; import org.apache.flink.runtime.concurrent.ScheduledExecutor; @@ -57,12 +58,15 @@ import org.apache.flink.runtime.testingUtils.TestingUtils; import org.apache.flink.runtime.testtasks.NoOpInvokable; import org.apache.flink.util.ExceptionUtils; +import org.apache.flink.util.FlinkException; import org.apache.flink.util.SerializedValue; import org.apache.flink.util.TestLogger; import org.junit.After; import org.junit.Test; +import javax.annotation.Nonnull; + import java.io.IOException; import java.net.InetAddress; import java.util.Iterator; @@ -86,15 +90,20 @@ import static org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.finishAllVertices; import static org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.switchToRunning; import static org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.waitUntilJobStatus; +import static org.hamcrest.Matchers.is; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; -import static org.mockito.Mockito.spy; +/** + * Tests the restart behaviour of the {@link ExecutionGraph}. + */ public class ExecutionGraphRestartTest extends TestLogger { - private final static int NUM_TASKS = 31; + private static final int NUM_TASKS = 31; private final ScheduledExecutorService executor = Executors.newScheduledThreadPool(4); @@ -113,9 +122,7 @@ public void testNoManualRestart() throws Exception { eg.getAllExecutionVertices().iterator().next().fail(new Exception("Test Exception")); - for (ExecutionVertex vertex : eg.getAllExecutionVertices()) { - vertex.getCurrentExecutionAttempt().cancelingComplete(); - } + completeCanceling(eg); assertEquals(JobStatus.FAILED, eg.getState()); @@ -125,9 +132,19 @@ public void testNoManualRestart() throws Exception { assertEquals(JobStatus.FAILED, eg.getState()); } + private void completeCanceling(ExecutionGraph eg) { + executeOperationForAllExecutions(eg, Execution::cancelingComplete); + } + + private void executeOperationForAllExecutions(ExecutionGraph eg, Consumer<Execution> operation) { + for (ExecutionVertex vertex : eg.getAllExecutionVertices()) { + operation.accept(vertex.getCurrentExecutionAttempt()); + } + } + @Test public void testRestartAutomatically() throws Exception { - RestartStrategy restartStrategy = new FixedDelayRestartStrategy(1, 1000); + RestartStrategy restartStrategy = new FixedDelayRestartStrategy(1, 0L); Tuple2<ExecutionGraph, Instance> executionGraphInstanceTuple = createExecutionGraph(restartStrategy); ExecutionGraph eg = executionGraphInstanceTuple.f0; @@ -207,23 +224,21 @@ public void testFailWhileRestarting() throws Exception { // Kill the instance and wait for the job to restart instance.markDead(); - Deadline deadline = TestingUtils.TESTING_DURATION().fromNow(); - - while (deadline.hasTimeLeft() && - executionGraph.getState() != JobStatus.RESTARTING) { - - Thread.sleep(100); - } + waitUntilJobStatus(executionGraph, JobStatus.RESTARTING, TestingUtils.TESTING_DURATION().toMillis()); assertEquals(JobStatus.RESTARTING, executionGraph.getState()); - // The restarting should not fail with an ordinary exception - executionGraph.failGlobal(new Exception("Test exception")); + // If we fail when being in RESTARTING, then we should try to restart again + final long globalModVersion = executionGraph.getGlobalModVersion(); + final Exception testException = new Exception("Test exception"); + executionGraph.failGlobal(testException); + assertNotEquals(globalModVersion, executionGraph.getGlobalModVersion()); assertEquals(JobStatus.RESTARTING, executionGraph.getState()); + assertEquals(testException, executionGraph.getFailureCause()); // we should have updated the failure cause // but it should fail when sending a SuppressRestartsException - executionGraph.failGlobal(new SuppressRestartsException(new Exception("Test exception"))); + executionGraph.failGlobal(new SuppressRestartsException(new Exception("Suppress restart exception"))); assertEquals(JobStatus.FAILED, executionGraph.getState()); @@ -254,9 +269,7 @@ public void testCancelWhileFailing() throws Exception { assertEquals(JobStatus.CANCELLING, graph.getState()); // let all tasks finish cancelling - for (ExecutionVertex vertex : graph.getVerticesTopologically().iterator().next().getTaskVertices()) { - vertex.getCurrentExecutionAttempt().cancelingComplete(); - } + completeCanceling(graph); assertEquals(JobStatus.CANCELED, graph.getState()); } @@ -267,11 +280,7 @@ public void testFailWhileCanceling() throws Exception { final ExecutionGraph graph = createExecutionGraph(restartStrategy).f0; assertEquals(JobStatus.RUNNING, graph.getState()); - - // switch all tasks to running - for (ExecutionVertex vertex : graph.getVerticesTopologically().iterator().next().getTaskVertices()) { - vertex.getCurrentExecutionAttempt().switchToRunning(); - } + switchAllTasksToRunning(graph); graph.cancel(); @@ -282,13 +291,15 @@ public void testFailWhileCanceling() throws Exception { assertEquals(JobStatus.FAILING, graph.getState()); // let all tasks finish cancelling - for (ExecutionVertex vertex : graph.getVerticesTopologically().iterator().next().getTaskVertices()) { - vertex.getCurrentExecutionAttempt().cancelingComplete(); - } + completeCanceling(graph); assertEquals(JobStatus.FAILED, graph.getState()); } + private void switchAllTasksToRunning(ExecutionGraph graph) { + executeOperationForAllExecutions(graph, Execution::switchToRunning); + } + @Test public void testNoRestartOnSuppressException() throws Exception { final ExecutionGraph eg = createExecutionGraph(new FixedDelayRestartStrategy(Integer.MAX_VALUE, 0)).f0; @@ -299,9 +310,7 @@ public void testNoRestartOnSuppressException() throws Exception { assertEquals(JobStatus.FAILING, eg.getState()); - for (ExecutionVertex vertex : eg.getAllExecutionVertices()) { - vertex.getCurrentExecutionAttempt().cancelingComplete(); - } + completeCanceling(eg); eg.waitUntilTerminal(); assertEquals(JobStatus.FAILED, eg.getState()); @@ -330,7 +339,7 @@ public void testFailingExecutionAfterRestart() throws Exception { JobVertex sender = ExecutionGraphTestUtils.createJobVertex("Task1", 1, NoOpInvokable.class); JobVertex receiver = ExecutionGraphTestUtils.createJobVertex("Task2", 1, NoOpInvokable.class); JobGraph jobGraph = new JobGraph("Pointwise job", sender, receiver); - ExecutionGraph eg = newExecutionGraph(new FixedDelayRestartStrategy(1, 1000), scheduler); + ExecutionGraph eg = newExecutionGraph(new FixedDelayRestartStrategy(1, 0L), scheduler); eg.attachJobGraph(jobGraph.getVerticesSortedTopologicallyFromSources()); assertEquals(JobStatus.CREATED, eg.getState()); @@ -766,6 +775,68 @@ public void testRestartWithSlotSharingAndNotEnoughResources() throws Exception { } } + /** + * Tests that the {@link ExecutionGraph} can handle concurrent failures while + * being in the RESTARTING state. + */ + @Test + public void testConcurrentFailureWhileRestarting() throws Exception { + final long timeout = 5000L; + + final CountDownLatch countDownLatch = new CountDownLatch(2); + final CountDownLatchRestartStrategy restartStrategy = new CountDownLatchRestartStrategy(countDownLatch); + final ExecutionGraph executionGraph = createSimpleExecutionGraph(restartStrategy, new TestingSlotProvider(ignored -> new CompletableFuture<>())); + + executionGraph.setQueuedSchedulingAllowed(true); + executionGraph.scheduleForExecution(); + + assertThat(executionGraph.getState(), is(JobStatus.RUNNING)); + + executionGraph.failGlobal(new FlinkException("Test exception")); + + executor.execute(() -> { + countDownLatch.countDown(); + try { + countDownLatch.await(); + } catch (InterruptedException e) { + ExceptionUtils.rethrow(e); + } + + executionGraph.failGlobal(new FlinkException("Concurrent exception")); + }); + + waitUntilJobStatus(executionGraph, JobStatus.RUNNING, timeout); + } + + private static final class CountDownLatchRestartStrategy implements RestartStrategy { + + private final CountDownLatch countDownLatch; + + private CountDownLatchRestartStrategy(CountDownLatch countDownLatch) { + this.countDownLatch = countDownLatch; + } + + @Override + public boolean canRestart() { + return true; + } + + @Override + public void restart(RestartCallback restarter, ScheduledExecutor executor) { + executor.execute(() -> { + countDownLatch.countDown(); + + try { + countDownLatch.await(); + } catch (InterruptedException e) { + ExceptionUtils.rethrow(e); + } + + restarter.triggerFullRecovery(); + }); + } + } + // ------------------------------------------------------------------------ // Utilities // ------------------------------------------------------------------------ @@ -846,14 +917,6 @@ public void run() { } private static Tuple2<ExecutionGraph, Instance> createExecutionGraph(RestartStrategy restartStrategy) throws Exception { - return createExecutionGraph(restartStrategy, false); - } - - private static Tuple2<ExecutionGraph, Instance> createSpyExecutionGraph(RestartStrategy restartStrategy) throws Exception { - return createExecutionGraph(restartStrategy, true); - } - - private static Tuple2<ExecutionGraph, Instance> createExecutionGraph(RestartStrategy restartStrategy, boolean isSpy) throws Exception { Instance instance = ExecutionGraphTestUtils.getInstance( new ActorTaskManagerGateway( new SimpleActorGateway(TestingUtils.directExecutionContext())), @@ -862,15 +925,7 @@ public void run() { Scheduler scheduler = new Scheduler(TestingUtils.defaultExecutionContext()); scheduler.newInstanceAvailable(instance); - JobVertex sender = ExecutionGraphTestUtils.createJobVertex("Task", NUM_TASKS, NoOpInvokable.class); - - JobGraph jobGraph = new JobGraph("Pointwise job", sender); - - ExecutionGraph eg = newExecutionGraph(restartStrategy, scheduler); - if (isSpy) { - eg = spy(eg); - } - eg.attachJobGraph(jobGraph.getVerticesSortedTopologicallyFromSources()); + ExecutionGraph eg = createSimpleExecutionGraph(restartStrategy, scheduler); assertEquals(JobStatus.CREATED, eg.getState()); @@ -879,7 +934,23 @@ public void run() { return new Tuple2<>(eg, instance); } - private static ExecutionGraph newExecutionGraph(RestartStrategy restartStrategy, Scheduler scheduler) throws IOException { + private static ExecutionGraph createSimpleExecutionGraph(RestartStrategy restartStrategy, SlotProvider slotProvider) throws IOException, JobException { + JobGraph jobGraph = createJobGraph(NUM_TASKS); + + ExecutionGraph eg = newExecutionGraph(restartStrategy, slotProvider); + eg.attachJobGraph(jobGraph.getVerticesSortedTopologicallyFromSources()); + + return eg; + } + + @Nonnull + private static JobGraph createJobGraph(int parallelism) { + JobVertex sender = ExecutionGraphTestUtils.createJobVertex("Task", parallelism, NoOpInvokable.class); + + return new JobGraph("Pointwise job", sender); + } + + private static ExecutionGraph newExecutionGraph(RestartStrategy restartStrategy, SlotProvider slotProvider) throws IOException { return new ExecutionGraph( TestingUtils.defaultExecutor(), TestingUtils.defaultExecutor(), @@ -889,7 +960,7 @@ private static ExecutionGraph newExecutionGraph(RestartStrategy restartStrategy, new SerializedValue<>(new ExecutionConfig()), AkkaUtils.getDefaultTimeout(), restartStrategy, - scheduler); + slotProvider); } private static void restartAfterFailure(ExecutionGraph eg, FiniteDuration timeout, boolean haltAfterRestart) throws InterruptedException, TimeoutException { @@ -922,8 +993,10 @@ private static void waitForAllResourcesToBeAssignedAfterAsyncRestart(ExecutionGr private static void waitForAsyncRestart(ExecutionGraph eg, FiniteDuration timeout) throws InterruptedException { Deadline deadline = timeout.fromNow(); + long waitingTime = 10L; while (deadline.hasTimeLeft() && eg.getState() != JobStatus.RUNNING) { - Thread.sleep(100); + Thread.sleep(waitingTime); + waitingTime = Math.min(waitingTime << 1, 100L); } } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java index 2da2ba49265..a397b6ca915 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java @@ -121,6 +121,9 @@ import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +/** + * Tests for the legacy {@link TaskManager}. + */ @SuppressWarnings("serial") public class TaskManagerTest extends TestLogger { @@ -133,7 +136,7 @@ private static ActorSystem system; - final static UUID leaderSessionID = UUID.randomUUID(); + static final UUID LEADER_SESSION_ID = UUID.randomUUID(); private TestingHighAvailabilityServices highAvailabilityServices; @@ -215,7 +218,6 @@ protected void run() { Collections.<InputGateDeploymentDescriptor>emptyList(), new ArrayList<PermanentBlobKey>(), Collections.emptyList(), 0); - new Within(d) { @Override @@ -261,8 +263,6 @@ else if (!(message instanceof TaskManagerMessages.Heartbeat)) { fail("Unexpected message: " + message); } } while (System.currentTimeMillis() < deadline); - - } }; } @@ -283,11 +283,11 @@ public void testJobSubmissionAndCanceling() { final ActorGateway testActorGateway = new AkkaActorGateway( getTestActor(), - leaderSessionID); + LEADER_SESSION_ID); try { - ActorRef jm = system.actorOf(Props.create(SimpleJobManager.class, leaderSessionID)); - jobManager = new AkkaActorGateway(jm, leaderSessionID); + ActorRef jm = system.actorOf(Props.create(SimpleJobManager.class, LEADER_SESSION_ID)); + jobManager = new AkkaActorGateway(jm, LEADER_SESSION_ID); highAvailabilityServices.setJobMasterLeaderRetriever( HighAvailabilityServices.DEFAULT_JOB_ID, @@ -404,7 +404,7 @@ protected void run() { } }; } - catch(Exception e) { + catch (Exception e) { e.printStackTrace(); fail(e.getMessage()); } @@ -424,11 +424,11 @@ public void testJobSubmissionAndStop() throws Exception { final ActorGateway testActorGateway = new AkkaActorGateway( getTestActor(), - leaderSessionID); + LEADER_SESSION_ID); try { - ActorRef jm = system.actorOf(Props.create(SimpleJobManager.class, leaderSessionID)); - jobManager = new AkkaActorGateway(jm, leaderSessionID); + ActorRef jm = system.actorOf(Props.create(SimpleJobManager.class, LEADER_SESSION_ID)); + jobManager = new AkkaActorGateway(jm, LEADER_SESSION_ID); highAvailabilityServices.setJobMasterLeaderRetriever( HighAvailabilityServices.DEFAULT_JOB_ID, @@ -554,11 +554,11 @@ public void testGateChannelEdgeMismatch() { final ActorGateway testActorGateway = new AkkaActorGateway( getTestActor(), - leaderSessionID); + LEADER_SESSION_ID); try { - ActorRef jm = system.actorOf(Props.create(SimpleJobManager.class, leaderSessionID)); - jobManager = new AkkaActorGateway(jm, leaderSessionID); + ActorRef jm = system.actorOf(Props.create(SimpleJobManager.class, LEADER_SESSION_ID)); + jobManager = new AkkaActorGateway(jm, LEADER_SESSION_ID); highAvailabilityServices.setJobMasterLeaderRetriever( HighAvailabilityServices.DEFAULT_JOB_ID, @@ -651,7 +651,7 @@ public void testRunJobWithForwardChannel() { final ActorGateway testActorGateway = new AkkaActorGateway( getTestActor(), - leaderSessionID); + LEADER_SESSION_ID); try { final JobID jid = new JobID(); @@ -661,8 +661,8 @@ public void testRunJobWithForwardChannel() { final ExecutionAttemptID eid1 = new ExecutionAttemptID(); final ExecutionAttemptID eid2 = new ExecutionAttemptID(); - ActorRef jm = system.actorOf(Props.create(new SimpleLookupJobManagerCreator(leaderSessionID))); - jobManager = new AkkaActorGateway(jm, leaderSessionID); + ActorRef jm = system.actorOf(Props.create(new SimpleLookupJobManagerCreator(LEADER_SESSION_ID))); + jobManager = new AkkaActorGateway(jm, LEADER_SESSION_ID); highAvailabilityServices.setJobMasterLeaderRetriever( HighAvailabilityServices.DEFAULT_JOB_ID, @@ -793,7 +793,7 @@ public void testCancellingDependentAndStateUpdateFails() { final ActorGateway testActorGateway = new AkkaActorGateway( getTestActor(), - leaderSessionID); + LEADER_SESSION_ID); try { final JobID jid = new JobID(); @@ -806,12 +806,12 @@ public void testCancellingDependentAndStateUpdateFails() { ActorRef jm = system.actorOf( Props.create( new SimpleLookupFailingUpdateJobManagerCreator( - leaderSessionID, + LEADER_SESSION_ID, eid2) ) ); - jobManager = new AkkaActorGateway(jm, leaderSessionID); + jobManager = new AkkaActorGateway(jm, LEADER_SESSION_ID); highAvailabilityServices.setJobMasterLeaderRetriever( HighAvailabilityServices.DEFAULT_JOB_ID, @@ -911,14 +911,14 @@ protected void run() { assertEquals(0, tasks.size()); } - catch(Exception e) { + catch (Exception e) { e.printStackTrace(); fail(e.getMessage()); } } }; } - catch(Exception e) { + catch (Exception e) { e.printStackTrace(); fail(e.getMessage()); } @@ -943,16 +943,16 @@ public void testRemotePartitionNotFound() throws Exception { final ActorGateway testActorGateway = new AkkaActorGateway( getTestActor(), - leaderSessionID); + LEADER_SESSION_ID); try { final IntermediateDataSetID resultId = new IntermediateDataSetID(); // Create the JM ActorRef jm = system.actorOf(Props.create( - new SimplePartitionStateLookupJobManagerCreator(leaderSessionID, getTestActor()))); + new SimplePartitionStateLookupJobManagerCreator(LEADER_SESSION_ID, getTestActor()))); - jobManager = new AkkaActorGateway(jm, leaderSessionID); + jobManager = new AkkaActorGateway(jm, LEADER_SESSION_ID); highAvailabilityServices.setJobMasterLeaderRetriever( HighAvailabilityServices.DEFAULT_JOB_ID, @@ -1017,12 +1017,12 @@ protected void run() { // The task should fail after repeated requests assertEquals(ExecutionState.FAILED, msg.getExecutionState()); Throwable t = msg.getError(ClassLoader.getSystemClassLoader()); - assertEquals("Thrown exception was not a PartitionNotFoundException: " + t.getMessage(), + assertEquals("Thrown exception was not a PartitionNotFoundException: " + t.getMessage(), PartitionNotFoundException.class, t.getClass()); } }; } - catch(Exception e) { + catch (Exception e) { e.printStackTrace(); fail(e.getMessage()); } @@ -1065,16 +1065,16 @@ public void testLocalPartitionNotFound() throws Exception { final ActorGateway testActorGateway = new AkkaActorGateway( getTestActor(), - leaderSessionID); + LEADER_SESSION_ID); try { final IntermediateDataSetID resultId = new IntermediateDataSetID(); // Create the JM ActorRef jm = system.actorOf(Props.create( - new SimplePartitionStateLookupJobManagerCreator(leaderSessionID, getTestActor()))); + new SimplePartitionStateLookupJobManagerCreator(LEADER_SESSION_ID, getTestActor()))); - jobManager = new AkkaActorGateway(jm, leaderSessionID); + jobManager = new AkkaActorGateway(jm, LEADER_SESSION_ID); highAvailabilityServices.setJobMasterLeaderRetriever( HighAvailabilityServices.DEFAULT_JOB_ID, @@ -1143,7 +1143,7 @@ protected void run() { } }; } - catch(Exception e) { + catch (Exception e) { e.printStackTrace(); fail(e.getMessage()); } @@ -1167,9 +1167,9 @@ public void testLogNotFoundHandling() throws Exception { // Create the JM ActorRef jm = system.actorOf(Props.create( - new SimplePartitionStateLookupJobManagerCreator(leaderSessionID, getTestActor()))); + new SimplePartitionStateLookupJobManagerCreator(LEADER_SESSION_ID, getTestActor()))); - jobManager = new AkkaActorGateway(jm, leaderSessionID); + jobManager = new AkkaActorGateway(jm, LEADER_SESSION_ID); final int dataPort = NetUtils.getAvailablePort(); Configuration config = new Configuration(); @@ -1209,7 +1209,8 @@ protected void run() { TestingUtils.stopActor(taskManager); TestingUtils.stopActor(jobManager); } - }};} + }}; + } // ------------------------------------------------------------------------ // Stack trace sample @@ -1555,7 +1556,7 @@ public void testTerminationOnFatalError() { * Test that a failing schedule or update consumers call leads to the failing of the respective * task. * - * IMPORTANT: We have to make sure that the invokable's cancel method is called, because only + * <p>IMPORTANT: We have to make sure that the invokable's cancel method is called, because only * then the future is completed. We do this by not eagerly deploy consumer tasks and requiring * the invokable to fill one memory segment. The completed memory segment will trigger the * scheduling of the downstream operator since it is in pipeline mode. After we've filled the @@ -1592,9 +1593,8 @@ public void testFailingScheduleOrUpdateConsumersMessage() throws Exception { Collections.<InputGateDeploymentDescriptor>emptyList(), new ArrayList<>(), Collections.emptyList(), 0); - - ActorRef jmActorRef = system.actorOf(Props.create(FailingScheduleOrUpdateConsumersJobManager.class, leaderSessionID), "jobmanager"); - ActorGateway jobManager = new AkkaActorGateway(jmActorRef, leaderSessionID); + ActorRef jmActorRef = system.actorOf(Props.create(FailingScheduleOrUpdateConsumersJobManager.class, LEADER_SESSION_ID), "jobmanager"); + ActorGateway jobManager = new AkkaActorGateway(jmActorRef, LEADER_SESSION_ID); highAvailabilityServices.setJobMasterLeaderRetriever( HighAvailabilityServices.DEFAULT_JOB_ID, @@ -1635,8 +1635,8 @@ public void testSubmitTaskFailure() throws Exception { try { - ActorRef jm = system.actorOf(Props.create(SimpleJobManager.class, leaderSessionID)); - jobManager = new AkkaActorGateway(jm, leaderSessionID); + ActorRef jm = system.actorOf(Props.create(SimpleJobManager.class, LEADER_SESSION_ID)); + jobManager = new AkkaActorGateway(jm, LEADER_SESSION_ID); highAvailabilityServices.setJobMasterLeaderRetriever( HighAvailabilityServices.DEFAULT_JOB_ID, @@ -1696,8 +1696,8 @@ public void testStopTaskFailure() throws Exception { try { final ExecutionAttemptID executionAttemptId = new ExecutionAttemptID(); - ActorRef jm = system.actorOf(Props.create(SimpleJobManager.class, leaderSessionID)); - jobManager = new AkkaActorGateway(jm, leaderSessionID); + ActorRef jm = system.actorOf(Props.create(SimpleJobManager.class, LEADER_SESSION_ID)); + jobManager = new AkkaActorGateway(jm, LEADER_SESSION_ID); highAvailabilityServices.setJobMasterLeaderRetriever( HighAvailabilityServices.DEFAULT_JOB_ID, @@ -1764,8 +1764,8 @@ public void testStackTraceSampleFailure() throws Exception { try { - ActorRef jm = system.actorOf(Props.create(SimpleJobManager.class, leaderSessionID)); - jobManager = new AkkaActorGateway(jm, leaderSessionID); + ActorRef jm = system.actorOf(Props.create(SimpleJobManager.class, LEADER_SESSION_ID)); + jobManager = new AkkaActorGateway(jm, LEADER_SESSION_ID); highAvailabilityServices.setJobMasterLeaderRetriever( HighAvailabilityServices.DEFAULT_JOB_ID, @@ -1813,8 +1813,8 @@ public void testUpdateTaskInputPartitionsFailure() throws Exception { final ExecutionAttemptID executionAttemptId = new ExecutionAttemptID(); - ActorRef jm = system.actorOf(Props.create(SimpleJobManager.class, leaderSessionID)); - jobManager = new AkkaActorGateway(jm, leaderSessionID); + ActorRef jm = system.actorOf(Props.create(SimpleJobManager.class, LEADER_SESSION_ID)); + jobManager = new AkkaActorGateway(jm, LEADER_SESSION_ID); highAvailabilityServices.setJobMasterLeaderRetriever( HighAvailabilityServices.DEFAULT_JOB_ID, @@ -1894,7 +1894,7 @@ public void handleMessage(Object message) throws Exception { ), self); } - else if(message instanceof TaskMessages.UpdateTaskExecutionState){ + else if (message instanceof TaskMessages.UpdateTaskExecutionState){ getSender().tell(true, getSelf()); } } @@ -1958,7 +1958,7 @@ public void handleMessage(Object message) throws Exception{ TaskMessages.UpdateTaskExecutionState updateMsg = (TaskMessages.UpdateTaskExecutionState) message; - if(validIDs.contains(updateMsg.taskExecutionState().getID())) { + if (validIDs.contains(updateMsg.taskExecutionState().getID())) { getSender().tell(true, getSelf()); } else { getSender().tell(false, getSelf()); @@ -2021,7 +2021,7 @@ public SimpleLookupFailingUpdateJobManagerCreator(UUID leaderSessionID, Executio validIDs = new HashSet<ExecutionAttemptID>(); - for(ExecutionAttemptID id : ids) { + for (ExecutionAttemptID id : ids) { this.validIDs.add(id); } } @@ -2049,9 +2049,9 @@ public SimplePartitionStateLookupJobManager create() throws Exception { return new SimplePartitionStateLookupJobManager(leaderSessionID, testActor); } } - + // -------------------------------------------------------------------------------------------- - + public static final class TestInvokableCorrect extends AbstractInvokable { public TestInvokableCorrect(Environment environment) { @@ -2061,7 +2061,7 @@ public TestInvokableCorrect(Environment environment) { @Override public void invoke() {} } - + public static class TestInvokableBlockingCancelable extends AbstractInvokable { public TestInvokableBlockingCancelable(Environment environment) { ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services