You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by dw...@apache.org on 2018/10/04 11:34:11 UTC
[flink] branch master updated (e061738 -> 4d5b7ef)
This is an automated email from the ASF dual-hosted git repository.
dwysakowicz pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from e061738 [hotfix][travis] Remove legacy references
new 93c87ea [FLINK-10354] Revert "[FLINK-6328] [chkPts] Don't add savepoints to CompletedCheckpointStore"
new 4d5b7ef [FLINK-10345][docs] Added note with warning about removing savepoints
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
docs/ops/state/savepoints.md | 5 +
.../runtime/checkpoint/CheckpointCoordinator.java | 37 +++----
.../checkpoint/CheckpointCoordinatorTest.java | 115 ++-------------------
3 files changed, 33 insertions(+), 124 deletions(-)
[flink] 01/02: [FLINK-10354] Revert "[FLINK-6328] [chkPts] Don't
add savepoints to CompletedCheckpointStore"
Posted by dw...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dwysakowicz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 93c87ea58e7f8bcf530be5a1bf7fa89a1e976f07
Author: Dawid Wysakowicz <dw...@apache.org>
AuthorDate: Mon Sep 17 12:04:11 2018 +0200
[FLINK-10354] Revert "[FLINK-6328] [chkPts] Don't add savepoints to CompletedCheckpointStore"
This reverts commit 6f570e7b6810e1645a4f7094f17ab9e8559fa139.
This closes #6704
---
.../runtime/checkpoint/CheckpointCoordinator.java | 37 +++----
.../checkpoint/CheckpointCoordinatorTest.java | 115 ++-------------------
2 files changed, 28 insertions(+), 124 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
index e936b24..57337b6 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
@@ -839,28 +839,22 @@ public class CheckpointCoordinator {
// the pending checkpoint must be discarded after the finalization
Preconditions.checkState(pendingCheckpoint.isDiscarded() && completedCheckpoint != null);
- // TODO: add savepoints to completed checkpoint store once FLINK-4815 has been completed
- if (!completedCheckpoint.getProperties().isSavepoint()) {
- try {
- completedCheckpointStore.addCheckpoint(completedCheckpoint);
- } catch (Exception exception) {
- // we failed to store the completed checkpoint. Let's clean up
- executor.execute(new Runnable() {
- @Override
- public void run() {
- try {
- completedCheckpoint.discardOnFailedStoring();
- } catch (Throwable t) {
- LOG.warn("Could not properly discard completed checkpoint {} of job {}.", completedCheckpoint.getCheckpointID(), job, t);
- }
+ try {
+ completedCheckpointStore.addCheckpoint(completedCheckpoint);
+ } catch (Exception exception) {
+ // we failed to store the completed checkpoint. Let's clean up
+ executor.execute(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ completedCheckpoint.discardOnFailedStoring();
+ } catch (Throwable t) {
+ LOG.warn("Could not properly discard completed checkpoint {}.", completedCheckpoint.getCheckpointID(), t);
}
- });
-
- throw new CheckpointException("Could not complete the pending checkpoint " + checkpointId + '.', exception);
- }
+ }
+ });
- // drop those pending checkpoints that are at prior to the completed one
- dropSubsumedCheckpoints(checkpointId);
+ throw new CheckpointException("Could not complete the pending checkpoint " + checkpointId + '.', exception);
}
} finally {
pendingCheckpoints.remove(checkpointId);
@@ -870,6 +864,9 @@ public class CheckpointCoordinator {
rememberRecentCheckpointId(checkpointId);
+ // drop those pending checkpoints that are at prior to the completed one
+ dropSubsumedCheckpoints(checkpointId);
+
// record the time when this was completed, to calculate
// the 'min delay between checkpoints'
lastCheckpointCompletionNanos = System.nanoTime();
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
index b113e12..3650f43 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
@@ -1494,8 +1494,8 @@ public class CheckpointCoordinatorTest extends TestLogger {
assertTrue(pending.isDiscarded());
assertTrue(savepointFuture.isDone());
- // the now the savepoint should be completed but not added to the completed checkpoint store
- assertEquals(0, coord.getNumberOfRetainedSuccessfulCheckpoints());
+ // the now we should have a completed checkpoint
+ assertEquals(1, coord.getNumberOfRetainedSuccessfulCheckpoints());
assertEquals(0, coord.getNumberOfPendingCheckpoints());
// validate that the relevant tasks got a confirmation message
@@ -1510,7 +1510,7 @@ public class CheckpointCoordinatorTest extends TestLogger {
verify(subtaskState2, times(1)).registerSharedStates(any(SharedStateRegistry.class));
}
- CompletedCheckpoint success = savepointFuture.get();
+ CompletedCheckpoint success = coord.getSuccessfulCheckpoints().get(0);
assertEquals(jid, success.getJobId());
assertEquals(timestamp, success.getTimestamp());
assertEquals(pending.getCheckpointId(), success.getCheckpointID());
@@ -1528,9 +1528,9 @@ public class CheckpointCoordinatorTest extends TestLogger {
coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jid, attemptID2, checkpointIdNew));
assertEquals(0, coord.getNumberOfPendingCheckpoints());
- assertEquals(0, coord.getNumberOfRetainedSuccessfulCheckpoints());
+ assertEquals(1, coord.getNumberOfRetainedSuccessfulCheckpoints());
- CompletedCheckpoint successNew = savepointFuture.get();
+ CompletedCheckpoint successNew = coord.getSuccessfulCheckpoints().get(0);
assertEquals(jid, successNew.getJobId());
assertEquals(timestampNew, successNew.getTimestamp());
assertEquals(checkpointIdNew, successNew.getCheckpointID());
@@ -1557,7 +1557,7 @@ public class CheckpointCoordinatorTest extends TestLogger {
* Triggers a savepoint and two checkpoints. The second checkpoint completes
* and subsumes the first checkpoint, but not the first savepoint. Then we
* trigger another checkpoint and savepoint. The 2nd savepoint completes and
- * does neither subsume the last checkpoint nor the first savepoint.
+ * subsumes the last checkpoint, but not the first savepoint.
*/
@Test
public void testSavepointsAreNotSubsumed() throws Exception {
@@ -1614,19 +1614,18 @@ public class CheckpointCoordinatorTest extends TestLogger {
assertFalse(savepointFuture1.isDone());
assertTrue(coord.triggerCheckpoint(timestamp + 3, false));
- long checkpointId3 = counter.getLast();
assertEquals(2, coord.getNumberOfPendingCheckpoints());
CompletableFuture<CompletedCheckpoint> savepointFuture2 = coord.triggerSavepoint(timestamp + 4, savepointDir);
long savepointId2 = counter.getLast();
assertEquals(3, coord.getNumberOfPendingCheckpoints());
- // 2nd savepoint should not subsume the last checkpoint and the 1st savepoint
+ // 2nd savepoint should subsume the last checkpoint, but not the 1st savepoint
coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jid, attemptID1, savepointId2));
coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jid, attemptID2, savepointId2));
- assertEquals(2, coord.getNumberOfPendingCheckpoints());
- assertEquals(1, coord.getNumberOfRetainedSuccessfulCheckpoints());
+ assertEquals(1, coord.getNumberOfPendingCheckpoints());
+ assertEquals(2, coord.getNumberOfRetainedSuccessfulCheckpoints());
assertFalse(coord.getPendingCheckpoints().get(savepointId1).isDiscarded());
assertFalse(savepointFuture1.isDone());
@@ -1636,15 +1635,9 @@ public class CheckpointCoordinatorTest extends TestLogger {
coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jid, attemptID1, savepointId1));
coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jid, attemptID2, savepointId1));
- assertEquals(1, coord.getNumberOfPendingCheckpoints());
- assertEquals(1, coord.getNumberOfRetainedSuccessfulCheckpoints());
- assertTrue(savepointFuture1.isDone());
-
- coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jid, attemptID1, checkpointId3));
- coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jid, attemptID2, checkpointId3));
-
assertEquals(0, coord.getNumberOfPendingCheckpoints());
- assertEquals(2, coord.getNumberOfRetainedSuccessfulCheckpoints());
+ assertEquals(3, coord.getNumberOfRetainedSuccessfulCheckpoints());
+ assertTrue(savepointFuture1.isDone());
}
private void testMaxConcurrentAttempts(int maxConcurrentAttempts) {
@@ -3467,92 +3460,6 @@ public class CheckpointCoordinatorTest extends TestLogger {
.reportRestoredCheckpoint(any(RestoredCheckpointStats.class));
}
- /**
- * FLINK-6328
- *
- * Tests that savepoints are not added to the {@link CompletedCheckpointStore} and,
- * thus, are not subject to job recovery. The reason that we don't want that (until
- * FLINK-4815 has been finished) is that the lifecycle of savepoints is not controlled
- * by the {@link CheckpointCoordinator}.
- */
- @Test
- public void testSavepointsAreNotAddedToCompletedCheckpointStore() throws Exception {
- final JobID jobId = new JobID();
- final ExecutionAttemptID executionAttemptId = new ExecutionAttemptID();
- final ExecutionVertex vertex1 = mockExecutionVertex(executionAttemptId);
- final CompletedCheckpointStore completedCheckpointStore = new StandaloneCompletedCheckpointStore(1);
- final long checkpointTimestamp1 = 1L;
- final long savepointTimestamp = 2L;
- final long checkpointTimestamp2 = 3L;
- final String savepointDir = tmpFolder.newFolder().getAbsolutePath();
-
- final StandaloneCheckpointIDCounter checkpointIDCounter = new StandaloneCheckpointIDCounter();
-
- CheckpointCoordinator checkpointCoordinator = new CheckpointCoordinator(
- jobId,
- 600000L,
- 600000L,
- 0L,
- Integer.MAX_VALUE,
- CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION,
- new ExecutionVertex[]{vertex1},
- new ExecutionVertex[]{vertex1},
- new ExecutionVertex[]{vertex1},
- checkpointIDCounter,
- completedCheckpointStore,
- new MemoryStateBackend(),
- Executors.directExecutor(),
- SharedStateRegistry.DEFAULT_FACTORY);
-
- // trigger a first checkpoint
- assertTrue(
- "Triggering of a checkpoint should work.",
- checkpointCoordinator.triggerCheckpoint(checkpointTimestamp1, false));
-
- assertTrue(0 == completedCheckpointStore.getNumberOfRetainedCheckpoints());
-
- // complete the 1st checkpoint
- checkpointCoordinator.receiveAcknowledgeMessage(
- new AcknowledgeCheckpoint(
- jobId,
- executionAttemptId,
- checkpointIDCounter.getLast()));
-
- // check that the checkpoint has been completed
- assertTrue(1 == completedCheckpointStore.getNumberOfRetainedCheckpoints());
-
- // trigger a savepoint --> this should not have any effect on the CompletedCheckpointStore
- CompletableFuture<CompletedCheckpoint> savepointFuture = checkpointCoordinator.triggerSavepoint(savepointTimestamp, savepointDir);
-
- checkpointCoordinator.receiveAcknowledgeMessage(
- new AcknowledgeCheckpoint(
- jobId,
- executionAttemptId,
- checkpointIDCounter.getLast()));
-
- // check that no errors occurred
- final CompletedCheckpoint savepoint = savepointFuture.get();
-
- assertFalse(
- "The savepoint should not have been added to the completed checkpoint store",
- savepoint.getCheckpointID() == completedCheckpointStore.getLatestCheckpoint().getCheckpointID());
-
- assertTrue(
- "Triggering of a checkpoint should work.",
- checkpointCoordinator.triggerCheckpoint(checkpointTimestamp2, false));
-
- // complete the 2nd checkpoint
- checkpointCoordinator.receiveAcknowledgeMessage(
- new AcknowledgeCheckpoint(
- jobId,
- executionAttemptId,
- checkpointIDCounter.getLast()));
-
- assertTrue(
- "The latest completed (proper) checkpoint should have been added to the completed checkpoint store.",
- completedCheckpointStore.getLatestCheckpoint().getCheckpointID() == checkpointIDCounter.getLast());
- }
-
@Test
public void testSharedStateRegistrationOnRestore() throws Exception {
[flink] 02/02: [FLINK-10345][docs] Added note with warning about
removing savepoints
Posted by dw...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dwysakowicz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 4d5b7efc3943fadae6b39eeb44b0411f450dce02
Author: Dawid Wysakowicz <dw...@apache.org>
AuthorDate: Thu Sep 20 11:19:31 2018 +0200
[FLINK-10345][docs] Added note with warning about removing savepoints
---
docs/ops/state/savepoints.md | 5 +++++
1 file changed, 5 insertions(+)
diff --git a/docs/ops/state/savepoints.md b/docs/ops/state/savepoints.md
index 26aaf27..be653f5 100644
--- a/docs/ops/state/savepoints.md
+++ b/docs/ops/state/savepoints.md
@@ -118,6 +118,11 @@ Please follow <a href="https://issues.apache.org/jira/browse/FLINK-5778">FLINK-5
Note that if you use the `MemoryStateBackend`, metadata *and* savepoint state will be stored in the `_metadata` file. Since it is self-contained, you may move the file and restore from any location.
+<div class="alert alert-warning">
+ <strong>Attention:</strong> It is discouraged to move or delete last savepoint of a running job, cause this might interfere with failure-recovery. Savepoints have side-effects on exactly-once sinks, therefore
+ to ensure exactly-once semantics, if there is no checkpoint after the last savepoint, the savepoint will be used for recovery.
+</div>
+
#### Trigger a Savepoint
{% highlight shell %}