Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/5548#discussion_r169960392
--- Diff:
flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphSchedulingTest.java
---
@@ -412,6 +417,54 @@ public void testEagerSchedulingWithSlotTimeout()
throws Exception {
verify(taskManager,
times(0)).submitTask(any(TaskDeploymentDescriptor.class), any(Time.class));
}
+ /**
+ * Tests that an ongoing scheduling operation does not fail the {@link
ExecutionGraph}
+ * if it gets concurrently cancelled
+ */
+ @Test
+ public void testSchedulingOperationCancellationWhenCancel() throws
Exception {
+ final JobVertex jobVertex = new JobVertex("NoOp JobVertex");
+ jobVertex.setInvokableClass(NoOpInvokable.class);
+ jobVertex.setParallelism(2);
+ final JobGraph jobGraph = new JobGraph(jobVertex);
+ jobGraph.setScheduleMode(ScheduleMode.EAGER);
+ jobGraph.setAllowQueuedScheduling(true);
+
+ final CompletableFuture<LogicalSlot> slotFuture1 = new
CompletableFuture<>();
+ final CompletableFuture<LogicalSlot> slotFuture2 = new
CompletableFuture<>();
+ final ProgrammedSlotProvider slotProvider = new
ProgrammedSlotProvider(2);
+ slotProvider.addSlots(jobVertex.getID(), new
CompletableFuture[]{slotFuture1, slotFuture2});
+ final ExecutionGraph executionGraph =
createExecutionGraph(jobGraph, slotProvider);
+
+ executionGraph.scheduleForExecution();
+
+ final CompletableFuture<?> releaseFuture = new
CompletableFuture<>();
+
+ final TestingLogicalSlot slot = new TestingLogicalSlot(
+ new LocalTaskManagerLocation(),
+ new SimpleAckingTaskManagerGateway(),
+ 0,
+ new AllocationID(),
+ new SlotRequestId(),
+ new SlotSharingGroupId(),
+ releaseFuture);
+ slotFuture1.complete(slot);
+
+ // cancel should change the state of all executions to CANCELLED
+ executionGraph.cancel();
+
+ // complete the now CANCELLED execution --> this should cause a
failure
+ slotFuture2.complete(new TestingLogicalSlot());
+
+ Thread.sleep(1L);
--- End diff --
Yes.
---
