[FLINK-19487][task] Fix calculation of checkpoinStartDelay for single…

… channel CheckpointBarrierAligner Previously for one single channel this metric was always zero
apache · Oct 13, 2020 · 1d43462 · 1d43462
1 parent fc5f49e
commit 1d43462
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 0 deletions.
diff --git a/...ng-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierAligner.java b/...ng-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierAligner.java
@@ -121,6 +121,7 @@ public void processBarrier(CheckpointBarrier receivedBarrier, InputChannelInfo c
 			if (barrierId > currentCheckpointId) {
 				// new checkpoint
 				currentCheckpointId = barrierId;
+				markAlignmentStartAndEnd(receivedBarrier.getTimestamp());
 				notifyCheckpoint(receivedBarrier);
 			}
 			return;

diff --git a/...t/java/org/apache/flink/streaming/runtime/io/AlternatingCheckpointBarrierHandlerTest.java b/...t/java/org/apache/flink/streaming/runtime/io/AlternatingCheckpointBarrierHandlerTest.java
@@ -157,6 +157,16 @@ public void testMetricsSingleChannel() throws Exception {
 			0L,
 			0L,
 			10_000_000L);
+
+		long checkpoint2CreationTime = System.currentTimeMillis() - 5;
+		sendBarrier(2, checkpoint2CreationTime, SAVEPOINT, gate, 0);
+		Thread.sleep(5);
+		assertMetrics(
+			gate.getCheckpointBarrierHandler(),
+			2L,
+			0L,
+			0L,
+			5_000_000L);
 	}
 
 	private void assertMetrics(