You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2018/02/01 15:46:37 UTC
[05/22] flink git commit: [hotfix] [checkpointing] Cleanup: Fix
Nullability and names in checkpoint stats.
[hotfix] [checkpointing] Cleanup: Fix Nullability and names in checkpoint stats.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/709c9ff6
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/709c9ff6
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/709c9ff6
Branch: refs/heads/master
Commit: 709c9ff6f7240132a25b5138f4541396a54e8fde
Parents: f8b41e4
Author: Stephan Ewen <se...@apache.org>
Authored: Tue Jan 30 15:53:46 2018 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Thu Feb 1 13:43:41 2018 +0100
----------------------------------------------------------------------
.../checkpoint/CompletedCheckpointStats.java | 21 +++++++++-----------
.../checkpoint/PendingCheckpointStats.java | 6 +++---
2 files changed, 12 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/709c9ff6/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStats.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStats.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStats.java
index f6b6aed..5ca355b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStats.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStats.java
@@ -21,6 +21,7 @@ package org.apache.flink.runtime.checkpoint;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import javax.annotation.Nullable;
+
import java.util.Map;
import static org.apache.flink.util.Preconditions.checkArgument;
@@ -35,7 +36,7 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
*/
public class CompletedCheckpointStats extends AbstractCheckpointStats {
- private static final long serialVersionUID = 138833868551861343L;
+ private static final long serialVersionUID = 138833868551861344L;
/** Total checkpoint state size over all subtasks. */
private final long stateSize;
@@ -46,9 +47,8 @@ public class CompletedCheckpointStats extends AbstractCheckpointStats {
/** The latest acknowledged subtask stats. */
private final SubtaskStateStats latestAcknowledgedSubtask;
- /** Optional external path if persisted externally. */
- @Nullable
- private final String externalPath;
+ /** The external pointer of the checkpoint. */
+ private final String externalPointer;
/** Flag indicating whether the checkpoint was discarded. */
private volatile boolean discarded;
@@ -65,7 +65,7 @@ public class CompletedCheckpointStats extends AbstractCheckpointStats {
* @param stateSize Total checkpoint state size over all subtasks.
* @param alignmentBuffered Buffered bytes during alignment over all subtasks.
* @param latestAcknowledgedSubtask The latest acknowledged subtask stats.
- * @param externalPath Optional external path if persisted externally.
+ * @param externalPointer Optional external path if persisted externally.
*/
CompletedCheckpointStats(
long checkpointId,
@@ -77,7 +77,7 @@ public class CompletedCheckpointStats extends AbstractCheckpointStats {
long stateSize,
long alignmentBuffered,
SubtaskStateStats latestAcknowledgedSubtask,
- @Nullable String externalPath) {
+ String externalPointer) {
super(checkpointId, triggerTimestamp, props, totalSubtaskCount, taskStats);
checkArgument(numAcknowledgedSubtasks == totalSubtaskCount, "Did not acknowledge all subtasks.");
@@ -85,7 +85,7 @@ public class CompletedCheckpointStats extends AbstractCheckpointStats {
this.stateSize = stateSize;
this.alignmentBuffered = alignmentBuffered;
this.latestAcknowledgedSubtask = checkNotNull(latestAcknowledgedSubtask);
- this.externalPath = externalPath;
+ this.externalPointer = externalPointer;
}
@Override
@@ -119,13 +119,10 @@ public class CompletedCheckpointStats extends AbstractCheckpointStats {
// ------------------------------------------------------------------------
/**
- * Returns the external path if this checkpoint was persisted externally.
- *
- * @return External path of this checkpoint or <code>null</code>.
+ * Returns the external pointer of this checkpoint.
*/
- @Nullable
public String getExternalPath() {
- return externalPath;
+ return externalPointer;
}
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/709c9ff6/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpointStats.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpointStats.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpointStats.java
index 59b404a..9b21747 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpointStats.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpointStats.java
@@ -140,10 +140,10 @@ public class PendingCheckpointStats extends AbstractCheckpointStats {
/**
* Reports a successfully completed pending checkpoint.
*
- * @param externalPath Optional external storage path if checkpoint was externalized.
+ * @param externalPointer Optional external storage path if checkpoint was externalized.
* @return Callback for the {@link CompletedCheckpoint} instance to notify about disposal.
*/
- CompletedCheckpointStats.DiscardCallback reportCompletedCheckpoint(@Nullable String externalPath) {
+ CompletedCheckpointStats.DiscardCallback reportCompletedCheckpoint(String externalPointer) {
CompletedCheckpointStats completed = new CompletedCheckpointStats(
checkpointId,
triggerTimestamp,
@@ -154,7 +154,7 @@ public class PendingCheckpointStats extends AbstractCheckpointStats {
currentStateSize,
currentAlignmentBuffered,
latestAcknowledgedSubtask,
- externalPath);
+ externalPointer);
trackerCallback.reportCompletedCheckpoint(completed);