First attempt at Full Checkpoint information on the UI

Author: ryanvanhuuksloot

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

@@ -182,6 +182,16 @@ public boolean isSavepoint() {
         return checkpointType.isSavepoint();
     }
 
+    /**
+     * Returns whether the checkpoint properties describe a full checkpoint.
+     *
+     * @return <code>true</code> if the properties describe a full checkpoint, <code>false</code>
+     *     otherwise.
+     */
+    public boolean isFullCheckpoint() {
+        return checkpointType.isFullCheckpoint();
+    }
+
     /**
      * Returns whether the checkpoint properties describe a synchronous savepoint/checkpoint.
      *

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

@@ -46,6 +46,10 @@ public boolean isSavepoint() {
         return false;
     }
 
+    public boolean isFullCheckpoint() {
+        return this == FULL_CHECKPOINT;
+    }
+
     public String getName() {
         return name;
     }

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

@@ -54,6 +54,10 @@ public boolean isSavepoint() {
         return true;
     }
 
+    public boolean isFullCheckpoint() {
+        return false;
+    }
+
     public boolean isSynchronous() {
         return postCheckpointAction != PostCheckpointAction.NONE;
     }

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

@@ -25,6 +25,8 @@ public interface SnapshotType extends Serializable {
 
     boolean isSavepoint();
 
+    boolean isFullCheckpoint();
+
     String getName();
 
     SharingFilesStrategy getSharingFilesStrategy();

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

@@ -88,6 +88,8 @@ public class CheckpointStatistics implements ResponseBody {
 
     public static final String FIELD_NAME_IS_SAVEPOINT = "is_savepoint";
 
+    public static final String FIELD_NAME_IS_FULL_CHECKPOINT = "is_full_checkpoint";
+
     public static final String FIELD_NAME_SAVEPOINT_FORMAT = "savepointFormat";
 
     public static final String FIELD_NAME_TRIGGER_TIMESTAMP = "trigger_timestamp";
@@ -129,6 +131,9 @@ public class CheckpointStatistics implements ResponseBody {
     @JsonProperty(FIELD_NAME_IS_SAVEPOINT)
     private final boolean savepoint;
 
+    @JsonProperty(FIELD_NAME_IS_FULL_CHECKPOINT)
+    private final boolean fullCheckpoint;
+
     @JsonProperty(FIELD_NAME_SAVEPOINT_FORMAT)
     @Nullable
     private final String savepointFormat;
@@ -175,6 +180,7 @@ private CheckpointStatistics(
             @JsonProperty(FIELD_NAME_ID) long id,
             @JsonProperty(FIELD_NAME_STATUS) CheckpointStatsStatus status,
             @JsonProperty(FIELD_NAME_IS_SAVEPOINT) boolean savepoint,
+            @JsonProperty(FIELD_NAME_IS_FULL_CHECKPOINT) boolean fullCheckpoint,
             @JsonProperty(FIELD_NAME_SAVEPOINT_FORMAT) String savepointFormat,
             @JsonProperty(FIELD_NAME_TRIGGER_TIMESTAMP) long triggerTimestamp,
             @JsonProperty(FIELD_NAME_LATEST_ACK_TIMESTAMP) long latestAckTimestamp,
@@ -193,6 +199,7 @@ private CheckpointStatistics(
         this.id = id;
         this.status = Preconditions.checkNotNull(status);
         this.savepoint = savepoint;
+        this.fullCheckpoint = fullCheckpoint;
         this.savepointFormat = savepointFormat;
         this.triggerTimestamp = triggerTimestamp;
         this.latestAckTimestamp = latestAckTimestamp;
@@ -220,6 +227,10 @@ public boolean isSavepoint() {
         return savepoint;
     }
 
+    public boolean isFullCheckpoint() {
+        return fullCheckpoint;
+    }
+
     public long getTriggerTimestamp() {
         return triggerTimestamp;
     }
@@ -268,6 +279,7 @@ public boolean equals(Object o) {
         CheckpointStatistics that = (CheckpointStatistics) o;
         return id == that.id
                 && savepoint == that.savepoint
+                && fullCheckpoint == that.fullCheckpoint
                 && Objects.equals(savepointFormat, that.savepointFormat)
                 && triggerTimestamp == that.triggerTimestamp
                 && latestAckTimestamp == that.latestAckTimestamp
@@ -289,6 +301,7 @@ public int hashCode() {
                 id,
                 status,
                 savepoint,
+                fullCheckpoint,
                 savepointFormat,
                 triggerTimestamp,
                 latestAckTimestamp,
@@ -352,6 +365,7 @@ public static CheckpointStatistics generateCheckpointStatistics(
                     completedCheckpointStats.getCheckpointId(),
                     completedCheckpointStats.getStatus(),
                     snapshotType.isSavepoint(),
+                    snapshotType.isFullCheckpoint(),
                     savepointFormat,
                     completedCheckpointStats.getTriggerTimestamp(),
                     completedCheckpointStats.getLatestAckTimestamp(),
@@ -377,6 +391,7 @@ public static CheckpointStatistics generateCheckpointStatistics(
                     failedCheckpointStats.getCheckpointId(),
                     failedCheckpointStats.getStatus(),
                     failedCheckpointStats.getProperties().isSavepoint(),
+                    failedCheckpointStats.getProperties().isFullCheckpoint(),
                     savepointFormat,
                     failedCheckpointStats.getTriggerTimestamp(),
                     failedCheckpointStats.getLatestAckTimestamp(),
@@ -402,6 +417,7 @@ public static CheckpointStatistics generateCheckpointStatistics(
                     pendingCheckpointStats.getCheckpointId(),
                     pendingCheckpointStats.getStatus(),
                     pendingCheckpointStats.getProperties().isSavepoint(),
+                    pendingCheckpointStats.getProperties().isFullCheckpoint(),
                     savepointFormat,
                     pendingCheckpointStats.getTriggerTimestamp(),
                     pendingCheckpointStats.getLatestAckTimestamp(),
@@ -474,6 +490,7 @@ public CompletedCheckpointStatistics(
                 @JsonProperty(FIELD_NAME_ID) long id,
                 @JsonProperty(FIELD_NAME_STATUS) CheckpointStatsStatus status,
                 @JsonProperty(FIELD_NAME_IS_SAVEPOINT) boolean savepoint,
+                @JsonProperty(FIELD_NAME_IS_FULL_CHECKPOINT) boolean fullCheckpoint,
                 @JsonProperty(FIELD_NAME_SAVEPOINT_FORMAT) String savepointFormat,
                 @JsonProperty(FIELD_NAME_TRIGGER_TIMESTAMP) long triggerTimestamp,
                 @JsonProperty(FIELD_NAME_LATEST_ACK_TIMESTAMP) long latestAckTimestamp,
@@ -495,6 +512,7 @@ public CompletedCheckpointStatistics(
                     id,
                     status,
                     savepoint,
+                    fullCheckpoint,
                     savepointFormat,
                     triggerTimestamp,
                     latestAckTimestamp,
@@ -562,6 +580,7 @@ public FailedCheckpointStatistics(
                 @JsonProperty(FIELD_NAME_ID) long id,
                 @JsonProperty(FIELD_NAME_STATUS) CheckpointStatsStatus status,
                 @JsonProperty(FIELD_NAME_IS_SAVEPOINT) boolean savepoint,
+                @JsonProperty(FIELD_NAME_IS_FULL_CHECKPOINT) boolean fullCheckpoint,
                 @JsonProperty(FIELD_NAME_SAVEPOINT_FORMAT) String savepointFormat,
                 @JsonProperty(FIELD_NAME_TRIGGER_TIMESTAMP) long triggerTimestamp,
                 @JsonProperty(FIELD_NAME_LATEST_ACK_TIMESTAMP) long latestAckTimestamp,
@@ -583,6 +602,7 @@ public FailedCheckpointStatistics(
                     id,
                     status,
                     savepoint,
+                    fullCheckpoint,
                     savepointFormat,
                     triggerTimestamp,
                     latestAckTimestamp,
@@ -640,6 +660,7 @@ public PendingCheckpointStatistics(
                 @JsonProperty(FIELD_NAME_ID) long id,
                 @JsonProperty(FIELD_NAME_STATUS) CheckpointStatsStatus status,
                 @JsonProperty(FIELD_NAME_IS_SAVEPOINT) boolean savepoint,
+                @JsonProperty(FIELD_NAME_IS_FULL_CHECKPOINT) boolean fullCheckpoint,
                 @JsonProperty(FIELD_NAME_SAVEPOINT_FORMAT) String savepointFormat,
                 @JsonProperty(FIELD_NAME_TRIGGER_TIMESTAMP) long triggerTimestamp,
                 @JsonProperty(FIELD_NAME_LATEST_ACK_TIMESTAMP) long latestAckTimestamp,
@@ -659,6 +680,7 @@ public PendingCheckpointStatistics(
                     id,
                     status,
                     savepoint,
+                    fullCheckpoint,
                     savepointFormat,
                     triggerTimestamp,
                     latestAckTimestamp,

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

@@ -78,6 +78,7 @@ protected CheckpointingStatistics getTestResponseInstance() throws Exception {
                         1L,
                         CheckpointStatsStatus.COMPLETED,
                         false,
+                        false,
                         null,
                         42L,
                         41L,
@@ -99,6 +100,7 @@ protected CheckpointingStatistics getTestResponseInstance() throws Exception {
                         2L,
                         CheckpointStatsStatus.COMPLETED,
                         true,
+                        false,
                         SavepointFormatType.CANONICAL.name(),
                         11L,
                         10L,
@@ -121,6 +123,7 @@ protected CheckpointingStatistics getTestResponseInstance() throws Exception {
                         3L,
                         CheckpointStatsStatus.FAILED,
                         false,
+                        false,
                         null,
                         5L,
                         10L,
@@ -145,6 +148,7 @@ protected CheckpointingStatistics getTestResponseInstance() throws Exception {
                         5L,
                         CheckpointStatsStatus.IN_PROGRESS,
                         false,
+                        false,
                         null,
                         42L,
                         41L,