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

…ntBarrierOvertakesCancellationBarrier in order to make the test more stable
apache · Jul 19, 2021 · 5af9790 · 5af9790
1 parent cf9c6bd
commit 5af9790
Showing 1 changed file with 21 additions and 7 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
@@ -357,13 +357,27 @@ 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,
+                greaterThan(beforeSecondCheckpointEndTime - afterSecondCheckpointStartTime));
+        assertThat(
+                alignmentTime,
+                lessThan(afterSecondCheckpointEndTime - beforeSecondCheckpointStartTime));
     }
 
     @Test