From 7d703b541dfa6f7f8c2cb99e611aa6fb27a41504 Mon Sep 17 00:00:00 2001
From: Yangze Guo <karmagyz@gmail.com>
Date: Sun, 25 Apr 2021 18:22:57 +0800
Subject: [PATCH] [FLINK-22074][test] Harden
 FineGrainedSlotManagerTest#testRequirementCheckOnlyTriggeredOnce in case
 deploying on a slot machine

---
 .../slotmanager/FineGrainedSlotManagerTest.java    | 14 +++++++++++++-
 1 file changed, 13 insertions(+), 1 deletion(-)

diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedSlotManagerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedSlotManagerTest.java
index 187093712bf8c..45d06a63e97a0 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedSlotManagerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedSlotManagerTest.java
@@ -54,6 +54,7 @@
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
+import static org.junit.Assume.assumeTrue;
 
 /** Tests of {@link FineGrainedSlotManager}. */
 public class FineGrainedSlotManagerTest extends FineGrainedSlotManagerTestBase {
@@ -567,7 +568,7 @@ public void testRequirementCheckOnlyTriggeredOnce() throws Exception {
                 final List<CompletableFuture<Void>> checkRequirementFutures = new ArrayList<>();
                 checkRequirementFutures.add(new CompletableFuture<>());
                 checkRequirementFutures.add(new CompletableFuture<>());
-                final long requirementCheckDelay = 20;
+                final long requirementCheckDelay = 50;
                 resourceAllocationStrategyBuilder.setTryFulfillRequirementsFunction(
                         (ignored1, ignored2) -> {
                             if (checkRequirementFutures.get(0).isDone()) {
@@ -588,6 +589,9 @@ public void testRequirementCheckOnlyTriggeredOnce() throws Exception {
                                     createResourceRequirementsForSingleSlot();
                             final TaskExecutorConnection taskExecutionConnection =
                                     createTaskExecutorConnection();
+                            final CompletableFuture<Void> registrationFuture =
+                                    new CompletableFuture<>();
+                            final long start = System.nanoTime();
                             runInMainThread(
                                     () -> {
                                         getSlotManager()
@@ -600,7 +604,15 @@ public void testRequirementCheckOnlyTriggeredOnce() throws Exception {
                                                         new SlotReport(),
                                                         DEFAULT_TOTAL_RESOURCE_PROFILE,
                                                         DEFAULT_SLOT_RESOURCE_PROFILE);
+                                        registrationFuture.complete(null);
                                     });
+
+                            assertFutureCompleteAndReturn(registrationFuture);
+                            final long registrationTime = (System.nanoTime() - start) / 1_000_000;
+                            assumeTrue(
+                                    "The time of process requirement and register task manager must not take longer than the requirement check delay. If it does, then this indicates a very slow machine.",
+                                    registrationTime < requirementCheckDelay);
+
                             assertFutureCompleteAndReturn(checkRequirementFutures.get(0));
                             assertFutureNotComplete(checkRequirementFutures.get(1));