[FLINK-20488][runtime][checkpoint] Show checkpoint type in the Web-UI (AC/UC) for each subtask

flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/subtask/job-checkpoints-subtask.component.html

@@ -91,6 +91,7 @@
         <th nzSortKey="alignment.processed" nzShowSort><strong>Processed (persisted) Data</strong></th>
         <th nzSortKey="alignment.duration" nzShowSort><strong>Alignment Duration</strong></th>
         <th nzSortKey="start_delay" nzShowSort><strong>Start Delay</strong></th>
+        <th nzSortKey="unaligned_checkpoint" nzShowSort><strong>Unaligned Checkpoint</strong></th>
       </tr>
     </thead>
     <tbody>
@@ -105,6 +106,7 @@
           <td>{{ subTask['alignment']['processed'] | humanizeBytes }} ({{ subTask['alignment']['persisted'] | humanizeBytes }})</td>
           <td>{{ subTask['alignment']['duration'] | humanizeDuration}}</td>
           <td>{{ subTask['start_delay'] | humanizeDuration}}</td>
+          <td>{{ subTask['unaligned_checkpoint']}}</td>
         </ng-container>
         <ng-container *ngIf="subTask['status'] == 'pending_or_failed'">
           <td colspan="7">n/a</td>

flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointMetrics.java

@@ -43,8 +43,11 @@ public class CheckpointMetrics implements Serializable {
 
     private final long checkpointStartDelayNanos;
 
+    /** Is the checkpoint completed as an unaligned checkpoint. */
+    private final boolean unalignedCheckpoint;
+
     public CheckpointMetrics() {
-        this(-1L, -1L, -1L, -1L, -1L, -1L);
+        this(-1L, -1L, -1L, -1L, -1L, -1L, false);
     }
 
     public CheckpointMetrics(
@@ -53,7 +56,8 @@ public class CheckpointMetrics implements Serializable {
             long alignmentDurationNanos,
             long syncDurationMillis,
             long asyncDurationMillis,
-            long checkpointStartDelayNanos) {
+            long checkpointStartDelayNanos,
+            boolean unalignedCheckpoint) {
 
         // these may be "-1", in case the values are unknown or not set
         checkArgument(bytesProcessedDuringAlignment >= -1);
@@ -69,6 +73,7 @@ public class CheckpointMetrics implements Serializable {
         this.syncDurationMillis = syncDurationMillis;
         this.asyncDurationMillis = asyncDurationMillis;
         this.checkpointStartDelayNanos = checkpointStartDelayNanos;
+        this.unalignedCheckpoint = unalignedCheckpoint;
     }
 
     public long getBytesProcessedDuringAlignment() {
@@ -95,6 +100,10 @@ public class CheckpointMetrics implements Serializable {
         return checkpointStartDelayNanos;
     }
 
+    public boolean getUnalignedCheckpoint() {
+        return unalignedCheckpoint;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {
@@ -111,7 +120,8 @@ public class CheckpointMetrics implements Serializable {
                 && alignmentDurationNanos == that.alignmentDurationNanos
                 && syncDurationMillis == that.syncDurationMillis
                 && asyncDurationMillis == that.asyncDurationMillis
-                && checkpointStartDelayNanos == that.checkpointStartDelayNanos;
+                && checkpointStartDelayNanos == that.checkpointStartDelayNanos
+                && unalignedCheckpoint == that.unalignedCheckpoint;
     }
 
     @Override
@@ -122,7 +132,8 @@ public class CheckpointMetrics implements Serializable {
                 alignmentDurationNanos,
                 syncDurationMillis,
                 asyncDurationMillis,
-                checkpointStartDelayNanos);
+                checkpointStartDelayNanos,
+                unalignedCheckpoint);
     }
 
     @Override
@@ -140,6 +151,8 @@ public class CheckpointMetrics implements Serializable {
                 + asyncDurationMillis
                 + ", checkpointStartDelayNanos="
                 + checkpointStartDelayNanos
+                + ", unalignedCheckpoint="
+                + unalignedCheckpoint
                 + '}';
     }
 }

flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointMetricsBuilder.java

@@ -40,6 +40,7 @@ public class CheckpointMetricsBuilder {
     private long syncDurationMillis = -1L;
     private long asyncDurationMillis = -1L;
     private long checkpointStartDelayNanos = -1L;
+    private boolean unalignedCheckpoint = false;
 
     public CheckpointMetricsBuilder setBytesProcessedDuringAlignment(
             long bytesProcessedDuringAlignment) {
@@ -116,6 +117,11 @@ public class CheckpointMetricsBuilder {
         return checkpointStartDelayNanos;
     }
 
+    public CheckpointMetricsBuilder setUnalignedCheckpoint(boolean unalignedCheckpoint) {
+        this.unalignedCheckpoint = unalignedCheckpoint;
+        return this;
+    }
+
     public CheckpointMetrics build() {
         return new CheckpointMetrics(
                 checkStateAndGet(bytesProcessedDuringAlignment),
@@ -123,6 +129,7 @@ public class CheckpointMetricsBuilder {
                 checkStateAndGet(alignmentDurationNanos),
                 syncDurationMillis,
                 asyncDurationMillis,
-                checkpointStartDelayNanos);
+                checkpointStartDelayNanos,
+                unalignedCheckpoint);
     }
 }

flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java

@@ -441,7 +441,8 @@ public class PendingCheckpoint implements Checkpoint {
                                 metrics.getBytesProcessedDuringAlignment(),
                                 metrics.getBytesPersistedDuringAlignment(),
                                 alignmentDurationMillis,
-                                checkpointStartDelayMillis);
+                                checkpointStartDelayMillis,
+                                metrics.getUnalignedCheckpoint());
 
                 statsCallback.reportSubtaskStats(vertex.getJobvertexId(), subtaskStateStats);
             }

flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SubtaskStateStats.java

@@ -58,6 +58,9 @@ public class SubtaskStateStats implements Serializable {
     /** Checkpoint start delay in milliseconds. */
     private final long checkpointStartDelay;
 
+    /** Is the checkpoint completed as an unaligned checkpoint. */
+    private final boolean unalignedCheckpoint;
+
     SubtaskStateStats(
             int subtaskIndex,
             long ackTimestamp,
@@ -67,7 +70,8 @@ public class SubtaskStateStats implements Serializable {
             long processedData,
             long persistedData,
             long alignmentDuration,
-            long checkpointStartDelay) {
+            long checkpointStartDelay,
+            boolean unalignedCheckpoint) {
 
         checkArgument(subtaskIndex >= 0, "Negative subtask index");
         this.subtaskIndex = subtaskIndex;
@@ -80,6 +84,7 @@ public class SubtaskStateStats implements Serializable {
         this.persistedData = persistedData;
         this.alignmentDuration = alignmentDuration;
         this.checkpointStartDelay = checkpointStartDelay;
+        this.unalignedCheckpoint = unalignedCheckpoint;
     }
 
     public int getSubtaskIndex() {
@@ -154,4 +159,8 @@ public class SubtaskStateStats implements Serializable {
     public long getCheckpointStartDelay() {
         return checkpointStartDelay;
     }
+
+    public boolean getUnalignedCheckpoint() {
+        return unalignedCheckpoint;
+    }
 }

flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/TaskCheckpointStatisticDetailsHandler.java

@@ -209,7 +209,8 @@ public class TaskCheckpointStatisticDetailsHandler
                                         subtask.getProcessedData(),
                                         subtask.getPersistedData(),
                                         subtask.getAlignmentDuration()),
-                                subtask.getCheckpointStartDelay()));
+                                subtask.getCheckpointStartDelay(),
+                                subtask.getUnalignedCheckpoint()));
             }
         }
 

flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/SubtaskCheckpointStatistics.java

@@ -106,6 +106,8 @@ public class SubtaskCheckpointStatistics {
 
         public static final String FIELD_NAME_START_DELAY = "start_delay";
 
+        public static final String FIELD_NAME_UNALIGNED_CHECKPOINT = "unaligned_checkpoint";
+
         @JsonProperty(FIELD_NAME_ACK_TIMESTAMP)
         private final long ackTimestamp;
 
@@ -124,6 +126,9 @@ public class SubtaskCheckpointStatistics {
         @JsonProperty(FIELD_NAME_START_DELAY)
         private final long startDelay;
 
+        @JsonProperty(FIELD_NAME_UNALIGNED_CHECKPOINT)
+        private boolean unalignedCheckpoint;
+
         @JsonCreator
         public CompletedSubtaskCheckpointStatistics(
                 @JsonProperty(FIELD_NAME_INDEX) int index,
@@ -132,7 +137,8 @@ public class SubtaskCheckpointStatistics {
                 @JsonProperty(FIELD_NAME_STATE_SIZE) long stateSize,
                 @JsonProperty(FIELD_NAME_CHECKPOINT_DURATION) CheckpointDuration checkpointDuration,
                 @JsonProperty(FIELD_NAME_ALIGNMENT) CheckpointAlignment alignment,
-                @JsonProperty(FIELD_NAME_START_DELAY) long startDelay) {
+                @JsonProperty(FIELD_NAME_START_DELAY) long startDelay,
+                @JsonProperty(FIELD_NAME_UNALIGNED_CHECKPOINT) boolean unalignedCheckpoint) {
             super(index, "completed");
             this.ackTimestamp = ackTimestamp;
             this.duration = duration;
@@ -140,6 +146,7 @@ public class SubtaskCheckpointStatistics {
             this.checkpointDuration = checkpointDuration;
             this.alignment = alignment;
             this.startDelay = startDelay;
+            this.unalignedCheckpoint = unalignedCheckpoint;
         }
 
         public long getAckTimestamp() {
@@ -166,6 +173,10 @@ public class SubtaskCheckpointStatistics {
             return startDelay;
         }
 
+        public boolean getUnalignedCheckpoint() {
+            return unalignedCheckpoint;
+        }
+
         @Override
         public boolean equals(Object o) {
             if (this == o) {
@@ -180,13 +191,20 @@ public class SubtaskCheckpointStatistics {
                     && stateSize == that.stateSize
                     && Objects.equals(checkpointDuration, that.checkpointDuration)
                     && Objects.equals(alignment, that.alignment)
-                    && startDelay == that.startDelay;
+                    && startDelay == that.startDelay
+                    && unalignedCheckpoint == that.unalignedCheckpoint;
         }
 
         @Override
         public int hashCode() {
             return Objects.hash(
-                    ackTimestamp, duration, stateSize, checkpointDuration, alignment, startDelay);
+                    ackTimestamp,
+                    duration,
+                    stateSize,
+                    checkpointDuration,
+                    alignment,
+                    startDelay,
+                    unalignedCheckpoint);
         }
 
         /** Duration of the checkpoint. */

flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointStatsTrackerTest.java

@@ -464,7 +464,8 @@ public class CheckpointStatsTrackerTest {
                         processedData,
                         persistedData,
                         ignored,
-                        ignored);
+                        ignored,
+                        false);
 
         assertTrue(pending.reportSubtaskStats(jobVertex.getJobVertexId(), subtaskStats));
 
@@ -548,6 +549,6 @@ public class CheckpointStatsTrackerTest {
     }
 
     private SubtaskStateStats createSubtaskStats(int index) {
-        return new SubtaskStateStats(index, 0, 0, 0, 0, 0, 0, 0, 0);
+        return new SubtaskStateStats(index, 0, 0, 0, 0, 0, 0, 0, 0, false);
     }
 }

flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointTest.java

@@ -379,7 +379,7 @@ public class CompletedCheckpointTest {
                         123129837912L,
                         42L,
                         44L,
-                        new SubtaskStateStats(123, 213123, 123123, 0, 0, 0, 0, 0, 0),
+                        new SubtaskStateStats(123, 213123, 123123, 0, 0, 0, 0, 0, 0, false),
                         null);
 
         CompletedCheckpointStats copy = CommonTestUtils.createCopySerializable(completed);

flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/PendingCheckpointStatsTest.java

@@ -294,6 +294,7 @@ public class PendingCheckpointStatsTest {
                 Integer.MAX_VALUE + (long) index,
                 Integer.MAX_VALUE + (long) index,
                 Integer.MAX_VALUE + (long) index,
-                Integer.MAX_VALUE + (long) index);
+                Integer.MAX_VALUE + (long) index,
+                false);
     }
 }

flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/SubtaskStateStatsTest.java

@@ -49,7 +49,8 @@ public class SubtaskStateStatsTest {
                         Integer.MAX_VALUE + 8L,
                         Integer.MAX_VALUE + 9L,
                         Integer.MAX_VALUE + 6L,
-                        Integer.MAX_VALUE + 7L);
+                        Integer.MAX_VALUE + 7L,
+                        false);
 
         stats = serialize ? CommonTestUtils.createCopySerializable(stats) : stats;
 

flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/TaskStateStatsTest.java

@@ -75,7 +75,8 @@ public class TaskStateStatsTest {
                             rand.nextInt(128),
                             rand.nextInt(128),
                             rand.nextInt(128),
-                            rand.nextInt(128));
+                            rand.nextInt(128),
+                            false);
 
             stateSize += subtasks[i].getStateSize();
             processedData += subtasks[i].getProcessedData();
@@ -94,7 +95,9 @@ public class TaskStateStatsTest {
             assertEquals(persistedData, taskStats.getPersistedDataStats());
         }
 
-        assertFalse(taskStats.reportSubtaskStats(new SubtaskStateStats(0, 0, 0, 0, 0, 0, 0, 0, 0)));
+        assertFalse(
+                taskStats.reportSubtaskStats(
+                        new SubtaskStateStats(0, 0, 0, 0, 0, 0, 0, 0, 0, false)));
 
         taskStats = serialize ? CommonTestUtils.createCopySerializable(taskStats) : taskStats;
 

flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/checkpoints/TaskCheckpointStatisticsWithSubtaskDetailsTest.java

@@ -64,7 +64,8 @@ public class TaskCheckpointStatisticsWithSubtaskDetailsTest
                                 .CheckpointDuration(1L, 2L),
                         new SubtaskCheckpointStatistics.CompletedSubtaskCheckpointStatistics
                                 .CheckpointAlignment(2L, 4L, 5L, 3L),
-                        42L));
+                        42L,
+                        true));
 
         return new TaskCheckpointStatisticsWithSubtaskDetails(
                 4L,

flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java

@@ -583,13 +583,15 @@ class SubtaskCheckpointCoordinatorImpl implements SubtaskCheckpointCoordinator {
         }
 
         LOG.debug(
-                "{} - finished synchronous part of checkpoint {}. Alignment duration: {} ms, snapshot duration {} ms",
+                "{} - finished synchronous part of checkpoint {}. Alignment duration: {} ms, snapshot duration {} ms, is unaligned checkpoint : {}",
                 taskName,
                 checkpointId,
                 checkpointMetrics.getAlignmentDurationNanosOrDefault() / 1_000_000,
-                checkpointMetrics.getSyncDurationMillis());
+                checkpointMetrics.getSyncDurationMillis(),
+                checkpointOptions.isUnalignedCheckpoint());
 
         checkpointMetrics.setSyncDurationMillis((System.nanoTime() - started) / 1_000_000);
+        checkpointMetrics.setUnalignedCheckpoint(checkpointOptions.isUnalignedCheckpoint());
         return true;
     }