fixup! fixup! [FLINK-38272][runtime] Fix unstable BatchJobRecoveryTest

Author: noorall

flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/TestingExecutionDeploymentTrackerWrapper.java

@@ -27,6 +27,7 @@
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 
+/** Testing implementation of the {@link ExecutionDeploymentTracker}. */
 public class TestingExecutionDeploymentTrackerWrapper implements ExecutionDeploymentTracker {
     private final ExecutionDeploymentTracker originalTracker;
     private final CompletableFuture<ExecutionAttemptID> taskDeploymentFuture;

flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptivebatch/BatchJobRecoveryTest.java

@@ -124,15 +124,13 @@
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
 
 import static org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.waitUntilExecutionVertexState;
 import static org.apache.flink.runtime.scheduler.DefaultSchedulerBuilder.createCustomParallelismDecider;
 import static org.apache.flink.runtime.util.JobVertexConnectionUtils.connectNewDataSetAsInput;
-import static org.apache.flink.util.Preconditions.checkArgument;
 import static org.apache.flink.util.Preconditions.checkState;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.fail;
@@ -170,7 +168,6 @@ public class BatchJobRecoveryTest {
     private static final int SOURCE_PARALLELISM = 5;
     private static final int MIDDLE_PARALLELISM = 5;
     private static final int DECIDED_SINK_PARALLELISM = 2;
-    private static final long TIMEOUT_MILLIS = 15000L;
     private static final JobVertexID SOURCE_ID = new JobVertexID();
     private static final JobVertexID MIDDLE_ID = new JobVertexID();
     private static final JobVertexID SINK_ID = new JobVertexID();
@@ -246,14 +243,14 @@ void testRecoverFromJMFailover() throws Exception {
 
         runInMainThread(scheduler::startScheduling);
 
-        waitUntilAllExecutionsDeployed(SOURCE_ID, scheduler, TIMEOUT_MILLIS);
+        waitUntilAllExecutionsDeployed(SOURCE_ID, scheduler);
         runInMainThread(
                 () -> {
                     // transition all sources to finished.
                     transitionExecutionsState(scheduler, ExecutionState.FINISHED, SOURCE_ID);
                 });
 
-        waitUntilAllExecutionsDeployed(MIDDLE_ID, scheduler, TIMEOUT_MILLIS);
+        waitUntilAllExecutionsDeployed(MIDDLE_ID, scheduler);
         runInMainThread(
                 () -> {
                     // transition all middle tasks to RUNNING state
@@ -349,14 +346,14 @@ void testJobVertexUnFinishedAndOperatorCoordinatorNotSupportBatchSnapshot() thro
 
         runInMainThread(scheduler::startScheduling);
 
-        waitUntilAllExecutionsDeployed(SOURCE_ID, scheduler, TIMEOUT_MILLIS);
+        waitUntilAllExecutionsDeployed(SOURCE_ID, scheduler);
         runInMainThread(
                 () -> {
                     // transition all sources to finished.
                     transitionExecutionsState(scheduler, ExecutionState.FINISHED, SOURCE_ID);
                 });
 
-        waitUntilAllExecutionsDeployed(MIDDLE_ID, scheduler, TIMEOUT_MILLIS);
+        waitUntilAllExecutionsDeployed(MIDDLE_ID, scheduler);
         runInMainThread(
                 () -> {
                     // transition first middle task to finished.
@@ -465,7 +462,7 @@ void testJobVertexFinishedAndOperatorCoordinatorNotSupportBatchSnapshotAndPartit
 
         runInMainThread(scheduler::startScheduling);
 
-        waitUntilAllExecutionsDeployed(SOURCE_ID, scheduler, TIMEOUT_MILLIS);
+        waitUntilAllExecutionsDeployed(SOURCE_ID, scheduler);
         runInMainThread(
                 () -> {
                     // transition all sources to finished.
@@ -510,9 +507,9 @@ void testJobVertexFinishedAndOperatorCoordinatorNotSupportBatchSnapshotAndPartit
             }
         }
 
-        waitUntilAllExecutionsDeployed(MIDDLE_ID, newScheduler, TIMEOUT_MILLIS);
+        waitUntilAllExecutionsDeployed(MIDDLE_ID, newScheduler);
 
-        waitUntilAllExecutionsDeployed(MIDDLE_ID, scheduler, TIMEOUT_MILLIS);
+        waitUntilAllExecutionsDeployed(MIDDLE_ID, scheduler);
         runInMainThread(
                 () -> {
                     // transition all middle tasks to running
@@ -553,7 +550,7 @@ void testRecoverFromJMFailoverAndPartitionsUnavailable() throws Exception {
 
         runInMainThread(scheduler::startScheduling);
 
-        waitUntilAllExecutionsDeployed(SOURCE_ID, scheduler, TIMEOUT_MILLIS);
+        waitUntilAllExecutionsDeployed(SOURCE_ID, scheduler);
         runInMainThread(
                 () -> {
                     // transition all sources to finished.
@@ -611,20 +608,20 @@ void testRecoverDecidedParallelismFromTheSameJobGraphInstance() throws Exception
 
         runInMainThread(scheduler::startScheduling);
 
-        waitUntilAllExecutionsDeployed(SOURCE_ID, scheduler, TIMEOUT_MILLIS);
+        waitUntilAllExecutionsDeployed(SOURCE_ID, scheduler);
         runInMainThread(
                 () -> {
                     // transition all sources to finished.
                     transitionExecutionsState(scheduler, ExecutionState.FINISHED, SOURCE_ID);
                 });
 
-        waitUntilAllExecutionsDeployed(MIDDLE_ID, scheduler, TIMEOUT_MILLIS);
+        waitUntilAllExecutionsDeployed(MIDDLE_ID, scheduler);
         runInMainThread(
                 () -> { // transition all middle tasks to finished.
                     transitionExecutionsState(scheduler, ExecutionState.FINISHED, MIDDLE_ID);
                 });
 
-        waitUntilAllExecutionsDeployed(SINK_ID, scheduler, TIMEOUT_MILLIS);
+        waitUntilAllExecutionsDeployed(SINK_ID, scheduler);
         runInMainThread(
                 () -> {
                     // transition all sinks to finished.
@@ -696,7 +693,7 @@ void testPartitionNotFoundTwiceAfterJMFailover() throws Exception {
                 });
 
         // transition all sources to finished.
-        waitUntilAllExecutionsDeployed(SOURCE_ID, scheduler, TIMEOUT_MILLIS);
+        waitUntilAllExecutionsDeployed(SOURCE_ID, scheduler);
         runInMainThread(
                 () -> transitionExecutionsState(scheduler, ExecutionState.FINISHED, SOURCE_ID));
 
@@ -1236,19 +1233,10 @@ public Optional<ResourceID> storesLocalResourcesOn() {
     }
 
     private void waitUntilAllExecutionsDeployed(
-            JobVertexID vertexId, AdaptiveBatchScheduler scheduler, long maxWaitMillis)
-            throws Exception {
-        checkArgument(maxWaitMillis >= 0);
-
-        // this is a poor implementation - we may want to improve it eventually
-        final long deadline =
-                maxWaitMillis == 0
-                        ? Long.MAX_VALUE
-                        : System.nanoTime() + (maxWaitMillis * 1_000_000);
-
+            JobVertexID vertexId, AdaptiveBatchScheduler scheduler) throws Exception {
         AtomicBoolean isAllExecutionDeployed = new AtomicBoolean(false);
 
-        while (System.nanoTime() < deadline && !isAllExecutionDeployed.get()) {
+        while (!isAllExecutionDeployed.get()) {
             runInMainThread(
                     () -> {
                         List<ExecutionAttemptID> attemptIds =
@@ -1266,13 +1254,7 @@ private void waitUntilAllExecutionsDeployed(
                             isAllExecutionDeployed.set(true);
                         }
                     });
-            try {
-                Thread.sleep(2);
-            } catch (InterruptedException ignored) {
-            }
-        }
-        if (!isAllExecutionDeployed.get()) {
-            throw new TimeoutException("Not all execution vertices deployed");
+            Thread.sleep(2);
         }
     }
 }