[FLINK-23419][streaming] Fixed the condition in testNextFirstCheckpoi…

…ntBarrierOvertakesCancellationBarrier in order to make the test more stable
apache · Jul 20, 2021 · 363f244 · 363f244
1 parent cf9c6bd
commit 363f244
Showing 1 changed file with 22 additions and 8 deletions.
diff --git a/...ava/src/test/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierTrackerTest.java b/...ava/src/test/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierTrackerTest.java
@@ -57,7 +57,6 @@
 import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 import static org.hamcrest.Matchers.is;
-import static org.hamcrest.Matchers.lessThan;
 import static org.hamcrest.Matchers.lessThanOrEqualTo;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
@@ -357,13 +356,28 @@ public void testNextFirstCheckpointBarrierOvertakesCancellationBarrier() throws
         ValidatingCheckpointHandler validator = new ValidatingCheckpointHandler();
         inputGate = createCheckpointedInputGate(2, sequence, validator);
 
-        for (BufferOrEvent boe : sequence) {
-            assertEquals(boe, inputGate.pollNext().get());
-            Thread.sleep(10);
-        }
-        long alignmentTime = validator.lastAlignmentDurationNanos.get() / 1_000_000;
-        assertThat(alignmentTime, greaterThan(0L));
-        assertThat(alignmentTime, lessThan(30L));
+        // Checkpoint 1
+        assertEquals(sequence[0], inputGate.pollNext().get());
+
+        long beforeSecondCheckpointStartTime = System.nanoTime();
+        // Checkpoint 2
+        assertEquals(sequence[1], inputGate.pollNext().get());
+        long afterSecondCheckpointStartTime = System.nanoTime();
+        // Cancellation checkpoint 1
+        assertEquals(sequence[2], inputGate.pollNext().get());
+        long beforeSecondCheckpointEndTime = System.nanoTime();
+        // Finishing checkpoint 2
+        assertEquals(sequence[3], inputGate.pollNext().get());
+        long afterSecondCheckpointEndTime = System.nanoTime();
+
+        long alignmentTime = validator.lastAlignmentDurationNanos.get();
+        assertThat(
+                alignmentTime,
+                greaterThanOrEqualTo(
+                        beforeSecondCheckpointEndTime - afterSecondCheckpointStartTime));
+        assertThat(
+                alignmentTime,
+                lessThanOrEqualTo(afterSecondCheckpointEndTime - beforeSecondCheckpointStartTime));
     }
 
     @Test