Reimplement the change without a sleep

Signed-off-by: davidradl <david_radley@uk.ibm.com>

Author: davidradl

flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphRestartTest.java

@@ -56,6 +56,7 @@
 import java.io.IOException;
 import java.util.Collections;
 import java.util.Iterator;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.function.Consumer;
 
@@ -198,6 +199,13 @@ private static ResourceID offerSlots(SlotPool slotPool, int numSlots) {
     @Test
     void testCancelWhileFailing() throws Exception {
         try (SlotPool slotPool = SlotPoolUtils.createDeclarativeSlotPoolBridge()) {
+            CompletableFuture<JobStatus> failingStatusFuture = new CompletableFuture<>();
+            JobStatusListener jobStatusListener =
+                    (jobId, newJobStatus, timestamp) -> {
+                        if (newJobStatus == JobStatus.FAILING) {
+                            failingStatusFuture.complete(newJobStatus);
+                        }
+                    };
             SchedulerBase scheduler =
                     new DefaultSchedulerBuilder(
                                     createJobGraph(),
@@ -207,6 +215,7 @@ void testCancelWhileFailing() throws Exception {
                                     createExecutionSlotAllocatorFactory(slotPool))
                             .setRestartBackoffTimeStrategy(
                                     new TestRestartBackoffTimeStrategy(false, Long.MAX_VALUE))
+                            .setJobStatusListener(jobStatusListener)
                             .build();
             ExecutionGraph graph = scheduler.getExecutionGraph();
 
@@ -220,8 +229,8 @@ void testCancelWhileFailing() throws Exception {
 
             scheduler.handleGlobalFailure(new Exception("test"));
 
-            // Give time for the failure to be processed
-            Thread.sleep(10);
+            // Wait for the state to transition to FAILING
+            failingStatusFuture.join();
 
             assertThat(graph.getState()).isEqualTo(JobStatus.FAILING);
 

flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/slowtaskdetector/ExecutionTimeBasedSlowTaskDetectorTest.java

@@ -72,11 +72,9 @@ void testNoFinishedTaskButRatioIsZero() throws Exception {
         final int parallelism = 3;
         final JobVertex jobVertex = createNoOpVertex(parallelism);
         final ExecutionGraph executionGraph = createExecutionGraph(jobVertex);
-
-        final ExecutionTimeBasedSlowTaskDetector slowTaskDetector = createSlowTaskDetector(0, 1, 0);
-
-        // Sleep to ensure that the tasks have some execution time
-        Thread.sleep(10);
+        // set lower bounds to 10ms. 0 can introduce intermittent unit test failures.
+        final ExecutionTimeBasedSlowTaskDetector slowTaskDetector =
+                createSlowTaskDetector(0, 1, 10);
 
         final Map<ExecutionVertexID, Collection<ExecutionAttemptID>> slowTasks =
                 slowTaskDetector.findSlowTasks(executionGraph);
@@ -251,8 +249,10 @@ void testBalancedInput() throws Exception {
                 DistributionPattern.ALL_TO_ALL,
                 ResultPartitionType.PIPELINED);
         final ExecutionGraph executionGraph = createExecutionGraph(jobVertex1, jobVertex2);
+        // with lower bounds 0, the test intermittently fails, as the expected 2 tasks had sometimes
+        // not yet started.
         final ExecutionTimeBasedSlowTaskDetector slowTaskDetector =
-                createSlowTaskDetector(0.3, 1, 0);
+                createSlowTaskDetector(0.3, 1, 10);
 
         final ExecutionVertex ev21 =
                 executionGraph.getJobVertex(jobVertex2.getID()).getTaskVertices()[0];
@@ -264,8 +264,6 @@ void testBalancedInput() throws Exception {
                 executionGraph.getJobVertex(jobVertex2.getID()).getTaskVertices()[2];
         ev23.setInputBytes(1024);
 
-        // Sleep to ensure that the 3 tasks do actually start
-        Thread.sleep(10);
         ev23.getCurrentExecutionAttempt().markFinished();
 
         final Map<ExecutionVertexID, Collection<ExecutionAttemptID>> slowTasks =