[FLINK-18955][Checkpointing] Add checkpoint path to job startup/resto…

…re message
deepfield · Aug 24, 2020 · 3b1b77e · 3b1b77e
1 parent 5487fcf
commit 3b1b77e
Show file tree

Hide file tree

Showing 3 changed files with 20 additions and 6 deletions.
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
@@ -1294,7 +1294,7 @@ private boolean restoreLatestCheckpointedStateInternal(
  }
  }
 
- LOG.info("Restoring job {} from latest valid checkpoint: {}.", job, latest);
+ LOG.info("Restoring job {} from {}.", job, latest);
 
  // re-assign the task states
  final Map<OperatorID, OperatorState> operatorStates = latest.getOperatorStates();

diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointType.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointType.java
@@ -24,24 +24,28 @@
 public enum CheckpointType {
 
  /** A checkpoint, full or incremental. */
- CHECKPOINT(false, false),
+ CHECKPOINT(false, false, "Checkpoint"),
 
  /** A regular savepoint. */
- SAVEPOINT(true, false),
+ SAVEPOINT(true, false, "Savepoint"),
 
  /** A savepoint taken while suspending/terminating the job. */
- SYNC_SAVEPOINT(true, true);
+ SYNC_SAVEPOINT(true, true, "Synchronous Savepoint");
 
  private final boolean isSavepoint;
 
  private final boolean isSynchronous;
 
+ private final String name;
+
  CheckpointType(
  final boolean isSavepoint,
- final boolean isSynchronous) {
+ final boolean isSynchronous,
+ final String name) {
 
  this.isSavepoint = isSavepoint;
  this.isSynchronous = isSynchronous;
+ this.name = name;
  }
 
  public boolean isSavepoint() {
@@ -51,4 +55,8 @@ public boolean isSavepoint() {
  public boolean isSynchronous() {
  return isSynchronous;
  }
+
+ public String getName() {
+ return name;
+ }
 }
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CompletedCheckpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CompletedCheckpoint.java
@@ -320,6 +320,12 @@ void setDiscardCallback(@Nullable CompletedCheckpointStats.DiscardCallback disca
 
  @Override
  public String toString() {
- return String.format("Checkpoint %d @ %d for %s", checkpointID, timestamp, job);
+ return String.format(
+ "%s %d @ %d for %s located at %s",
+ props.getCheckpointType().getName(),
+ checkpointID,
+ timestamp,
+ job,
+ externalPointer);
  }
 }