You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by uc...@apache.org on 2016/01/18 17:49:48 UTC
[1/2] flink git commit: [FLINK-3244] [runtime] Add debug log messages
to SavepointCoordinator
Repository: flink
Updated Branches:
refs/heads/master 39a4b5a9a -> 7d7d8b68e
[FLINK-3244] [runtime] Add debug log messages to SavepointCoordinator
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/7d7d8b68
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/7d7d8b68
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/7d7d8b68
Branch: refs/heads/master
Commit: 7d7d8b68ea74ec814a98b9798189d662da1b4e96
Parents: a5bba3c
Author: Ufuk Celebi <uc...@apache.org>
Authored: Mon Jan 18 14:09:17 2016 +0100
Committer: Ufuk Celebi <uc...@apache.org>
Committed: Mon Jan 18 17:49:08 2016 +0100
----------------------------------------------------------------------
.../checkpoint/SavepointCoordinator.java | 21 ++++++++++++++++----
1 file changed, 17 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/7d7d8b68/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SavepointCoordinator.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SavepointCoordinator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SavepointCoordinator.java
index e154e5e..6ce6502 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SavepointCoordinator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SavepointCoordinator.java
@@ -31,14 +31,12 @@ import org.apache.flink.runtime.instance.ActorGateway;
import org.apache.flink.runtime.instance.AkkaActorGateway;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobmanager.RecoveryMode;
-import org.apache.flink.types.IntValue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.concurrent.Future;
import scala.concurrent.Promise;
import java.util.Collections;
-import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
@@ -198,14 +196,29 @@ public class SavepointCoordinator extends CheckpointCoordinator {
LOG.info("Savepoint: {}@{}", checkpoint.getCheckpointID(), checkpoint.getTimestamp());
// Set the initial state of all tasks
+ LOG.debug("Rolling back individual operators.");
for (StateForTask state : checkpoint.getStates()) {
+ LOG.debug("Rolling back subtask {} of operator {}.",
+ state.getSubtask(), state.getOperatorId());
+
ExecutionJobVertex vertex = tasks.get(state.getOperatorId());
if (vertex == null) {
String msg = String.format("Failed to rollback to savepoint %s. " +
"Cannot map old state for task %s to the new program. " +
- "This indicates that the program has been changed after " +
- "the savepoint.", savepoint, state.getOperatorId());
+ "This indicates that the program has been changed in a " +
+ "non-compatible way after the savepoint.", savepoint,
+ state.getOperatorId());
+ throw new IllegalStateException(msg);
+ }
+
+ if (state.getSubtask() >= vertex.getParallelism()) {
+ String msg = String.format("Failed to rollback to savepoint %s. " +
+ "Parallelism mismatch between savepoint state and new program. " +
+ "Cannot map subtask %d of operator %s to new program with " +
+ "parallelism %d. This indicates that the program has been changed " +
+ "in a non-compatible way after the savepoint.", savepoint,
+ state.getSubtask(), state.getOperatorId(), vertex.getParallelism());
throw new IllegalStateException(msg);
}
[2/2] flink git commit: [FLINK-3250] [runtime] Remove too strict
parallelism check in SavepointCoordinator
Posted by uc...@apache.org.
[FLINK-3250] [runtime] Remove too strict parallelism check in SavepointCoordinator
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/a5bba3c6
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/a5bba3c6
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/a5bba3c6
Branch: refs/heads/master
Commit: a5bba3c6d89d858fa99ac9439a792e4524190b4a
Parents: 39a4b5a
Author: Ufuk Celebi <uc...@apache.org>
Authored: Mon Jan 18 13:56:03 2016 +0100
Committer: Ufuk Celebi <uc...@apache.org>
Committed: Mon Jan 18 17:49:08 2016 +0100
----------------------------------------------------------------------
.../checkpoint/SavepointCoordinator.java | 23 --------------------
.../checkpoint/SavepointCoordinatorTest.java | 18 ++-------------
2 files changed, 2 insertions(+), 39 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/a5bba3c6/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SavepointCoordinator.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SavepointCoordinator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SavepointCoordinator.java
index 2089aa2..e154e5e 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SavepointCoordinator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SavepointCoordinator.java
@@ -197,11 +197,6 @@ public class SavepointCoordinator extends CheckpointCoordinator {
LOG.info("Savepoint: {}@{}", checkpoint.getCheckpointID(), checkpoint.getTimestamp());
- // Sanity check to ensure that the parallelism has not been changed. If the tasks have
- // lower parallelism than the savepoint tasks, this will be noticed during reset, but
- // the other way around (higher parallelism than savepoint tasks) might go unnoticed.
- Map<JobVertexID, IntValue> vertexParallelism = new HashMap<>();
-
// Set the initial state of all tasks
for (StateForTask state : checkpoint.getStates()) {
ExecutionJobVertex vertex = tasks.get(state.getOperatorId());
@@ -214,28 +209,10 @@ public class SavepointCoordinator extends CheckpointCoordinator {
throw new IllegalStateException(msg);
}
- IntValue parallelism = vertexParallelism.get(vertex.getJobVertexId());
-
- if (parallelism == null) {
- parallelism = new IntValue(vertex.getParallelism());
- vertexParallelism.put(vertex.getJobVertexId(), parallelism);
- }
-
Execution exec = vertex.getTaskVertices()[state.getSubtask()]
.getCurrentExecutionAttempt();
exec.setInitialState(state.getState(), recoveryTimestamp);
-
- parallelism.setValue(parallelism.getValue() - 1);
- }
-
- // If the parallelism matches, each count is 0
- for (IntValue parallelism : vertexParallelism.values()) {
- if (parallelism.getValue() != 0) {
- throw new IllegalStateException("Parallelism mismatch between savepoint " +
- "tasks and new program. This indicates that the program has been " +
- "changed after the savepoint.");
- }
}
// Reset the checkpoint ID counter
http://git-wip-us.apache.org/repos/asf/flink/blob/a5bba3c6/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/SavepointCoordinatorTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/SavepointCoordinatorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/SavepointCoordinatorTest.java
index 4dc9c5c..4f9ae60 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/SavepointCoordinatorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/SavepointCoordinatorTest.java
@@ -259,22 +259,8 @@ public class SavepointCoordinatorTest {
String savepointPath = Await.result(savepointPathFuture, FiniteDuration.Zero());
assertNotNull(savepointPath);
- // Change parallelism higher than original (subtask without matching state)
- for (int i = 0; i < jobVertices.length; i++) {
- jobVertices[i] = mockExecutionJobVertex(jobId, jobVertices[i].getJobVertexId(), 8);
- }
-
- try {
- // Rollback
- coordinator.restoreSavepoint(
- createExecutionJobVertexMap(jobVertices),
- savepointPath);
- fail("Did not throw expected Exception after rollback with parallelism mismatch.");
- }
- catch (Exception ignored) {
- }
-
- // Change parallelism lower than original (state without matching subtask)
+ // Change parallelism lower than original (state without matching subtask). The
+ // other way around (subtask without matching state) is OK.
for (int i = 0; i < jobVertices.length; i++) {
jobVertices[i] = mockExecutionJobVertex(jobId, jobVertices[i].getJobVertexId(), 2);
}