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 2016/08/24 21:12:52 UTC
[3/6] flink git commit: [FLINK-4417] [checkpoints] Checkpoints are
subsumed by CheckpointID not, by timestamp
[FLINK-4417] [checkpoints] Checkpoints are subsumed by CheckpointID not, by timestamp
This closes #2407
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/4e9d1775
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/4e9d1775
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/4e9d1775
Branch: refs/heads/master
Commit: 4e9d1775b5514c87981c78d55323cc2b17361867
Parents: 4da40bc
Author: Ramkrishna <ra...@intel.com>
Authored: Tue Aug 23 21:53:31 2016 +0530
Committer: Stephan Ewen <se...@apache.org>
Committed: Wed Aug 24 19:56:17 2016 +0200
----------------------------------------------------------------------
.../flink/runtime/checkpoint/CheckpointCoordinator.java | 7 ++++---
.../org/apache/flink/runtime/jobmanager/JobManager.scala | 2 +-
2 files changed, 5 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/4e9d1775/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
----------------------------------------------------------------------
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
index 2c0e63b..b710324 100644
--- 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
@@ -671,7 +671,7 @@ public class CheckpointCoordinator {
pendingCheckpoints.remove(checkpointId);
rememberRecentCheckpointId(checkpointId);
- dropSubsumedCheckpoints(completed.getTimestamp());
+ dropSubsumedCheckpoints(completed.getCheckpointID());
triggerQueuedRequests();
}
@@ -726,12 +726,13 @@ public class CheckpointCoordinator {
recentPendingCheckpoints.addLast(id);
}
- private void dropSubsumedCheckpoints(long timestamp) throws Exception {
+ private void dropSubsumedCheckpoints(long checkpointId) throws Exception {
Iterator<Map.Entry<Long, PendingCheckpoint>> entries = pendingCheckpoints.entrySet().iterator();
while (entries.hasNext()) {
PendingCheckpoint p = entries.next().getValue();
- if (p.getCheckpointTimestamp() <= timestamp && p.canBeSubsumed()) {
+ // remove all pending checkpoints that are lesser than the current completed checkpoint
+ if (p.getCheckpointId() < checkpointId && p.canBeSubsumed()) {
rememberRecentCheckpointId(p.getCheckpointId());
p.abortSubsumed();
entries.remove();
http://git-wip-us.apache.org/repos/asf/flink/blob/4e9d1775/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
index d172a2b..34fed3f 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
@@ -1418,7 +1418,7 @@ class JobManager(
if (checkpointCoordinator != null) {
future {
try {
- if (checkpointCoordinator.receiveDeclineMessage(declineMessage)) {
+ if (!checkpointCoordinator.receiveDeclineMessage(declineMessage)) {
log.info("Received message for non-existing checkpoint " +
declineMessage.getCheckpointId)
}