[FLINK-37008] [runtime-web] Flink UI should show the type of checkpoint (full vs incremental)

flink-runtime-web/src/test/resources/rest_api_v1.snapshot

@@ -811,6 +811,9 @@
                 "additionalProperties" : {
                   "type" : "integer"
                 }
+              },
+              "pending-operators" : {
+                "type" : "integer"
               }
             }
           }

flink-runtime-web/web-dashboard/src/app/interfaces/job-checkpoint.ts

@@ -39,6 +39,7 @@ export interface Checkpoint {
       id: number;
       status: string;
       is_savepoint: boolean;
+      is_full: boolean;
       trigger_timestamp: number;
       latest_ack_timestamp: number;
       state_size: number;
@@ -55,6 +56,7 @@ export interface Checkpoint {
       id: number;
       restore_timestamp: number;
       is_savepoint: boolean;
+      is_full: boolean;
       external_path: string;
     };
     history: CheckpointHistory;
@@ -65,6 +67,7 @@ export interface CheckpointHistory {
   id: number;
   status: string;
   is_savepoint: boolean;
+  is_full: boolean;
   trigger_timestamp: number;
   latest_ack_timestamp: number;
   state_size: number;
@@ -91,6 +94,7 @@ export interface CheckpointCompletedStatistics {
   id: number;
   status: string;
   is_savepoint: boolean;
+  is_full: boolean;
   trigger_timestamp: number;
   latest_ack_timestamp: number;
   state_size: number;
@@ -142,6 +146,7 @@ export interface CheckpointDetail {
   id: number;
   status: string;
   is_savepoint: boolean;
+  is_full: boolean;
   savepointFormat: string;
   trigger_timestamp: number;
   latest_ack_timestamp: number;

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

@@ -25,7 +25,7 @@
     {{ checkPointDetail?.discarded || '-' }}
     <nz-divider nzType="vertical"></nz-divider>
     <strong>Checkpoint Type:</strong>
-    {{ checkPointType }}
+    {{ checkPointType }} ({{ checkPointDetail?.is_full ? 'full' : 'incremental' }})
     <ng-container *ngIf="checkPointDetail?.savepointFormat">
       <nz-divider nzType="vertical"></nz-divider>
       <strong>Savepoint Format:</strong>

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

@@ -1099,7 +1099,8 @@ private Optional<CheckpointTriggerRequest> chooseRequestToExecute(
     // Returns true if the checkpoint is successfully completed, false otherwise.
     private boolean maybeCompleteCheckpoint(PendingCheckpoint checkpoint) {
         synchronized (lock) {
-            if (checkpoint.isFullyAcknowledged()) {
+            if (checkpoint.isFullyAcknowledged()
+                    && checkpoint.getCheckpointType().isTypeResolved()) {
                 try {
                     // we need to check inside the lock for being shutdown as well,
                     // otherwise we get races and invalid error log messages.
@@ -2449,4 +2450,20 @@ private void reportFinishedTasks(
     private PendingCheckpointStats getStatsCallback(PendingCheckpoint pendingCheckpoint) {
         return statsTracker.getPendingCheckpointStats(pendingCheckpoint.getCheckpointID());
     }
+
+    /**
+     * Reports the resolution of a checkpoint type from a subtask.
+     *
+     * @param checkpointId ID of the checkpoint
+     * @param executionAttemptID ID of the reporting task
+     * @param isFull Whether the checkpoint was determined to be full
+     */
+    public void reportCheckpointTypeResolution(
+            long checkpointId, ExecutionAttemptID executionAttemptID, boolean isFull) {
+
+        PendingCheckpoint checkpoint = pendingCheckpoints.get(checkpointId);
+        if (checkpoint != null) {
+            checkpoint.getCheckpointType().resolveType(isFull);
+        }
+    }
 }

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 describes a full checkpoint, <code>false
+     *     </code> otherwise.
+     */
+    public boolean isFull() {
+        return checkpointType.isFull();
+    }
+
     /**
      * Returns whether the checkpoint properties describe a synchronous savepoint/checkpoint.
      *
@@ -283,6 +293,39 @@ public String toString() {
                     false, // Retain on suspension
                     false);
 
+    private static final CheckpointProperties FULL_CHECKPOINT_NEVER_RETAINED =
+            new CheckpointProperties(
+                    false,
+                    CheckpointType.FULL_CHECKPOINT,
+                    true,
+                    true, // Delete on success
+                    true, // Delete on cancellation
+                    true, // Delete on failure
+                    true, // Delete on suspension
+                    false);
+
+    private static final CheckpointProperties FULL_CHECKPOINT_RETAINED_ON_FAILURE =
+            new CheckpointProperties(
+                    false,
+                    CheckpointType.FULL_CHECKPOINT,
+                    true,
+                    true, // Delete on success
+                    true, // Delete on cancellation
+                    false, // Retain on failure
+                    true, // Delete on suspension
+                    false);
+
+    private static final CheckpointProperties FULL_CHECKPOINT_RETAINED_ON_CANCELLATION =
+            new CheckpointProperties(
+                    false,
+                    CheckpointType.FULL_CHECKPOINT,
+                    true,
+                    true, // Delete on success
+                    false, // Retain on cancellation
+                    false, // Retain on failure
+                    false, // Retain on suspension
+                    false);
+
     /**
      * Creates the checkpoint properties for a (manually triggered) savepoint.
      *
@@ -362,4 +405,27 @@ public static CheckpointProperties forCheckpoint(CheckpointRetentionPolicy polic
                 throw new IllegalArgumentException("unknown policy: " + policy);
         }
     }
+
+    /**
+     * Creates the checkpoint properties for a full checkpoint.
+     *
+     * <p>Checkpoints may be queued in case too many other checkpoints are currently happening. They
+     * are garbage collected automatically, except when the owning job terminates in state {@link
+     * JobStatus#FAILED}. The user is required to configure the clean up behaviour on job
+     * cancellation.
+     *
+     * @return Checkpoint properties for an external full checkpoint.
+     */
+    public static CheckpointProperties forFullCheckpoint(CheckpointRetentionPolicy policy) {
+        switch (policy) {
+            case NEVER_RETAIN_AFTER_TERMINATION:
+                return FULL_CHECKPOINT_NEVER_RETAINED;
+            case RETAIN_ON_FAILURE:
+                return FULL_CHECKPOINT_RETAINED_ON_FAILURE;
+            case RETAIN_ON_CANCELLATION:
+                return FULL_CHECKPOINT_RETAINED_ON_CANCELLATION;
+            default:
+                throw new IllegalArgumentException("unknown policy: " + policy);
+        }
+    }
 }

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

@@ -25,21 +25,32 @@
 /** The type of checkpoint to perform. */
 @Internal
 public final class CheckpointType implements SnapshotType {
+    private volatile boolean typeResolved = false;
+    private volatile boolean isFull = false;
 
     /** A checkpoint, full or incremental. */
     public static final CheckpointType CHECKPOINT =
             new CheckpointType("Checkpoint", SharingFilesStrategy.FORWARD_BACKWARD);
 
+    /** A checkpoint that is explicitly marked as full */
     public static final CheckpointType FULL_CHECKPOINT =
-            new CheckpointType("Full Checkpoint", SharingFilesStrategy.FORWARD);
+            new CheckpointType("Full Checkpoint", SharingFilesStrategy.FORWARD, true);
 
     private final String name;
-
     private final SharingFilesStrategy sharingFilesStrategy;
 
-    private CheckpointType(final String name, SharingFilesStrategy sharingFilesStrategy) {
+    private CheckpointType(String name, SharingFilesStrategy sharingFilesStrategy) {
+        this(name, sharingFilesStrategy, false);
+    }
+
+    private CheckpointType(
+            String name, SharingFilesStrategy sharingFilesStrategy, boolean preResolvedAsFull) {
         this.name = name;
         this.sharingFilesStrategy = sharingFilesStrategy;
+        if (preResolvedAsFull) {
+            this.isFull = true;
+            this.typeResolved = true;
+        }
     }
 
     public boolean isSavepoint() {
@@ -54,6 +65,21 @@ public SharingFilesStrategy getSharingFilesStrategy() {
         return sharingFilesStrategy;
     }
 
+    public boolean isFull() {
+        return typeResolved && isFull;
+    }
+
+    public void resolveType(boolean isFull) {
+        if (!typeResolved) {
+            this.isFull = isFull;
+            this.typeResolved = true;
+        }
+    }
+
+    public boolean isTypeResolved() {
+        return typeResolved;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {
@@ -63,12 +89,15 @@ public boolean equals(Object o) {
             return false;
         }
         CheckpointType type = (CheckpointType) o;
-        return name.equals(type.name) && sharingFilesStrategy == type.sharingFilesStrategy;
+        return typeResolved == type.typeResolved
+                && isFull == type.isFull
+                && name.equals(type.name)
+                && sharingFilesStrategy == type.sharingFilesStrategy;
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(name, sharingFilesStrategy);
+        return Objects.hash(name, sharingFilesStrategy, typeResolved, isFull);
     }
 
     @Override
@@ -79,6 +108,10 @@ public String toString() {
                 + '\''
                 + ", sharingFilesStrategy="
                 + sharingFilesStrategy
+                + ", typeResolved="
+                + typeResolved
+                + ", isFull="
+                + isFull
                 + '}';
     }
 }

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

@@ -25,6 +25,10 @@ public interface SnapshotType extends Serializable {
 
     boolean isSavepoint();
 
+    default boolean isFull() {
+        return getSharingFilesStrategy() != SharingFilesStrategy.FORWARD_BACKWARD;
+    }
+
     String getName();
 
     SharingFilesStrategy getSharingFilesStrategy();