You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gy...@apache.org on 2015/05/19 22:23:49 UTC
[2/6] flink git commit: [streaming] Discard method added to state
handle
[streaming] Discard method added to state handle
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/59bee4ad
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/59bee4ad
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/59bee4ad
Branch: refs/heads/master
Commit: 59bee4ad825d7252c791e0376ea2504602134fe7
Parents: 48a2720
Author: Gyula Fora <gy...@apache.org>
Authored: Thu May 14 11:49:30 2015 +0200
Committer: Gyula Fora <gy...@apache.org>
Committed: Tue May 19 18:25:53 2015 +0200
----------------------------------------------------------------------
.../checkpoint/CheckpointCoordinator.java | 25 ++++++++++++++------
.../runtime/checkpoint/PendingCheckpoint.java | 23 +++++++++++-------
.../flink/runtime/checkpoint/StateForTask.java | 18 ++++++++++++--
.../checkpoint/SuccessfulCheckpoint.java | 9 ++++++-
.../runtime/executiongraph/ExecutionGraph.java | 2 +-
.../runtime/state/ByteStreamStateHandle.java | 12 ++++++++++
.../flink/runtime/state/FileStateHandle.java | 15 +++++++++++-
.../flink/runtime/state/LocalStateHandle.java | 6 ++++-
.../apache/flink/runtime/state/StateHandle.java | 7 ++++++
.../apache/flink/runtime/taskmanager/Task.java | 4 +---
.../checkpoint/CheckpointCoordinatorTest.java | 16 +++++++------
.../checkpoint/CheckpointStateRestoreTest.java | 10 ++++----
.../messages/CheckpointMessagesTest.java | 6 ++++-
.../streaming/runtime/tasks/StreamTask.java | 7 +++---
14 files changed, 120 insertions(+), 40 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/59bee4ad/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 be0fcf2..e4abbf8 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
@@ -97,6 +97,8 @@ public class CheckpointCoordinator {
private ActorRef jobStatusListener;
+ private ClassLoader userClassLoader;
+
private boolean shutdown;
// --------------------------------------------------------------------------------------------
@@ -104,7 +106,8 @@ public class CheckpointCoordinator {
public CheckpointCoordinator(JobID job, int numSuccessfulCheckpointsToRetain, long checkpointTimeout,
ExecutionVertex[] tasksToTrigger,
ExecutionVertex[] tasksToWaitFor,
- ExecutionVertex[] tasksToCommitTo) {
+ ExecutionVertex[] tasksToCommitTo,
+ ClassLoader userClassLoader) {
// some sanity checks
if (job == null || tasksToTrigger == null ||
@@ -127,6 +130,7 @@ public class CheckpointCoordinator {
this.pendingCheckpoints = new LinkedHashMap<Long, PendingCheckpoint>();
this.completedCheckpoints = new ArrayDeque<SuccessfulCheckpoint>(numSuccessfulCheckpointsToRetain + 1);
this.recentPendingCheckpoints = new ArrayDeque<Long>(NUM_GHOST_CHECKPOINT_IDS);
+ this.userClassLoader = userClassLoader;
timer = new Timer("Checkpoint Timer", true);
}
@@ -166,13 +170,13 @@ public class CheckpointCoordinator {
// clear and discard all pending checkpoints
for (PendingCheckpoint pending : pendingCheckpoints.values()) {
- pending.discard();
+ pending.discard(userClassLoader, true);
}
pendingCheckpoints.clear();
// clean and discard all successful checkpoints
for (SuccessfulCheckpoint checkpoint : completedCheckpoints) {
- checkpoint.dispose();
+ checkpoint.dispose(userClassLoader);
}
completedCheckpoints.clear();
}
@@ -253,7 +257,9 @@ public class CheckpointCoordinator {
// note that checkpoint completion discards the pending checkpoint object
if (!checkpoint.isDiscarded()) {
LOG.info("Checkpoint " + checkpointID + " expired before completing.");
- checkpoint.discard();
+
+ checkpoint.discard(userClassLoader, true);
+
pendingCheckpoints.remove(checkpointID);
rememberRecentCheckpointId(checkpointID);
}
@@ -288,7 +294,10 @@ public class CheckpointCoordinator {
LOG.warn("Failed to trigger checkpoint (" + numUnsuccessful + " consecutive failed attempts so far)", t);
synchronized (lock) {
- pendingCheckpoints.remove(checkpointID);
+ PendingCheckpoint checkpoint = pendingCheckpoints.remove(checkpointID);
+ if (checkpoint != null && !checkpoint.isDiscarded()) {
+ checkpoint.discard(userClassLoader, true);
+ }
}
return false;
@@ -325,7 +334,7 @@ public class CheckpointCoordinator {
completed = checkpoint.toCompletedCheckpoint();
completedCheckpoints.addLast(completed);
if (completedCheckpoints.size() > numSuccessfulCheckpointsToRetain) {
- completedCheckpoints.removeFirst();
+ completedCheckpoints.removeFirst().dispose(userClassLoader);;
}
pendingCheckpoints.remove(checkpointId);
rememberRecentCheckpointId(checkpointId);
@@ -383,7 +392,9 @@ public class CheckpointCoordinator {
PendingCheckpoint p = entries.next().getValue();
if (p.getCheckpointTimestamp() < timestamp) {
rememberRecentCheckpointId(p.getCheckpointId());
- p.discard();
+
+ p.discard(userClassLoader, true);
+
entries.remove();
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/59bee4ad/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java
index f25bff9..9ea3b6f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java
@@ -18,16 +18,16 @@
package org.apache.flink.runtime.checkpoint;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
import org.apache.flink.runtime.executiongraph.ExecutionVertex;
import org.apache.flink.runtime.state.StateHandle;
import org.apache.flink.runtime.util.SerializedValue;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-
/**
* A pending checkpoint is a checkpoint that has been started, but has not been
* acknowledged by all tasks that need to acknowledge it. Once all tasks have
@@ -37,7 +37,7 @@ import java.util.Map;
* state handles always as serialized values, never as actual values.</p>
*/
public class PendingCheckpoint {
-
+
private final Object lock = new Object();
private final JobID jobId;
@@ -53,7 +53,7 @@ public class PendingCheckpoint {
private int numAcknowledgedTasks;
private boolean discarded;
-
+
// --------------------------------------------------------------------------------------------
public PendingCheckpoint(JobID jobId, long checkpointId, long checkpointTimestamp,
@@ -117,7 +117,8 @@ public class PendingCheckpoint {
if (notYetAcknowledgedTasks.isEmpty()) {
SuccessfulCheckpoint completed = new SuccessfulCheckpoint(jobId, checkpointId,
checkpointTimestamp, new ArrayList<StateForTask>(collectedStates));
- discard();
+ discard(null, false);
+
return completed;
}
else {
@@ -148,11 +149,17 @@ public class PendingCheckpoint {
/**
* Discards the pending checkpoint, releasing all held resources.
+ * @throws Exception
*/
- public void discard() {
+ public void discard(ClassLoader userClassLoader, boolean discardStateHandle) {
synchronized (lock) {
discarded = true;
numAcknowledgedTasks = -1;
+ if (discardStateHandle) {
+ for (StateForTask state : collectedStates) {
+ state.discard(userClassLoader);
+ }
+ }
collectedStates.clear();
notYetAcknowledgedTasks.clear();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/59bee4ad/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StateForTask.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StateForTask.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StateForTask.java
index 26b3eb7..073b22f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StateForTask.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StateForTask.java
@@ -18,9 +18,13 @@
package org.apache.flink.runtime.checkpoint;
+import java.io.IOException;
+
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.state.StateHandle;
import org.apache.flink.runtime.util.SerializedValue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* Simple bean to describe the state belonging to a parallel operator.
@@ -33,6 +37,8 @@ import org.apache.flink.runtime.util.SerializedValue;
* the respective classloader.
*/
public class StateForTask {
+
+ private static final Logger LOG = LoggerFactory.getLogger(StateForTask.class);
/** The state of the parallel operator */
private final SerializedValue<StateHandle<?>> state;
@@ -42,9 +48,9 @@ public class StateForTask {
/** The index of the parallel subtask */
private final int subtask;
-
+
public StateForTask(SerializedValue<StateHandle<?>> state, JobVertexID operatorId, int subtask) {
- if (state == null || operatorId == null || subtask < 0) {
+ if (state == null || operatorId == null || subtask < 0) {
throw new IllegalArgumentException();
}
@@ -66,6 +72,14 @@ public class StateForTask {
public int getSubtask() {
return subtask;
}
+
+ public void discard(ClassLoader userClassLoader) {
+ try {
+ state.deserializeValue(userClassLoader).discardState();
+ } catch (Exception e) {
+ LOG.warn("Failed to discard checkpoint state: " + this, e);
+ }
+ }
// --------------------------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/59bee4ad/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SuccessfulCheckpoint.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SuccessfulCheckpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SuccessfulCheckpoint.java
index cd7efba..3f77138 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SuccessfulCheckpoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SuccessfulCheckpoint.java
@@ -19,6 +19,8 @@
package org.apache.flink.runtime.checkpoint;
import org.apache.flink.api.common.JobID;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.util.List;
@@ -28,6 +30,8 @@ import java.util.List;
*/
public class SuccessfulCheckpoint {
+ private static final Logger LOG = LoggerFactory.getLogger(SuccessfulCheckpoint.class);
+
private final JobID job;
private final long checkpointID;
@@ -62,7 +66,10 @@ public class SuccessfulCheckpoint {
// --------------------------------------------------------------------------------------------
- public void dispose() {
+ public void dispose(ClassLoader userClassLoader) {
+ for(StateForTask state: states){
+ state.discard(userClassLoader);
+ }
states.clear();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/59bee4ad/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
index a041f86..73e827f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
@@ -307,7 +307,7 @@ public class ExecutionGraph implements Serializable {
// create the coordinator that triggers and commits checkpoints and holds the state
snapshotCheckpointsEnabled = true;
checkpointCoordinator = new CheckpointCoordinator(jobID, NUMBER_OF_SUCCESSFUL_CHECKPOINTS_TO_RETAIN,
- checkpointTimeout, tasksToTrigger, tasksToWaitFor, tasksToCommitTo);
+ checkpointTimeout, tasksToTrigger, tasksToWaitFor, tasksToCommitTo, userClassLoader);
// the periodic checkpoint scheduler is activated and deactivated as a result of
// job status changes (running -> on, all other states -> off)
http://git-wip-us.apache.org/repos/asf/flink/blob/59bee4ad/flink-runtime/src/main/java/org/apache/flink/runtime/state/ByteStreamStateHandle.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/ByteStreamStateHandle.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/ByteStreamStateHandle.java
index 36598d9..a202a83 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/ByteStreamStateHandle.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/ByteStreamStateHandle.java
@@ -25,6 +25,12 @@ import java.io.ObjectOutputStream;
import java.io.OutputStream;
import java.io.Serializable;
+/**
+ * Statehandle that writes/reads the contents of the serializable checkpointed
+ * state to the provided input and outputstreams using default java
+ * serialization.
+ *
+ */
public abstract class ByteStreamStateHandle implements StateHandle<Serializable> {
private static final long serialVersionUID = -962025800339325828L;
@@ -35,8 +41,14 @@ public abstract class ByteStreamStateHandle implements StateHandle<Serializable>
this.state = state;
}
+ /**
+ * The state will be written to the stream returned by this method.
+ */
protected abstract OutputStream getOutputStream() throws Exception;
+ /**
+ * The state will be read from the stream returned by this method.
+ */
protected abstract InputStream getInputStream() throws Exception;
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/59bee4ad/flink-runtime/src/main/java/org/apache/flink/runtime/state/FileStateHandle.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/FileStateHandle.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/FileStateHandle.java
index 9d614d8..956bd9d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/FileStateHandle.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/FileStateHandle.java
@@ -31,11 +31,18 @@ import org.apache.flink.util.StringUtils;
import scala.util.Random;
+/**
+ * Statehandle that writes the checkpointed state to a random file in the
+ * provided checkpoint directory. Any Flink supported File system can be used
+ * but it is advised to use a filesystem that is persistent in case of node
+ * failures, such as HDFS or Tachyon.
+ *
+ */
public class FileStateHandle extends ByteStreamStateHandle {
private static final long serialVersionUID = 1L;
- String pathString;
+ private String pathString;
public FileStateHandle(Serializable state, String folder) throws IOException {
super(state);
@@ -55,4 +62,10 @@ public class FileStateHandle extends ByteStreamStateHandle {
new Random().nextBytes(bytes);
return StringUtils.byteToHexString(bytes);
}
+
+ @Override
+ public void discardState() throws Exception {
+ FileSystem.get(new URI(pathString)).delete(new Path(pathString), false);
+ }
+
}
http://git-wip-us.apache.org/repos/asf/flink/blob/59bee4ad/flink-runtime/src/main/java/org/apache/flink/runtime/state/LocalStateHandle.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/LocalStateHandle.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/LocalStateHandle.java
index fa0c515..e8fe768 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/LocalStateHandle.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/LocalStateHandle.java
@@ -21,7 +21,7 @@ package org.apache.flink.runtime.state;
import java.io.Serializable;
/**
- * A StateHandle that includes a map of operator states directly.
+ * A StateHandle that includes the operator states directly.
*/
public class LocalStateHandle implements StateHandle<Serializable> {
@@ -37,4 +37,8 @@ public class LocalStateHandle implements StateHandle<Serializable> {
public Serializable getState() {
return state;
}
+
+ @Override
+ public void discardState() throws Exception {
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/59bee4ad/flink-runtime/src/main/java/org/apache/flink/runtime/state/StateHandle.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/StateHandle.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/StateHandle.java
index 409383c..c1342b8 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/StateHandle.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/StateHandle.java
@@ -34,4 +34,11 @@ public interface StateHandle<T> extends Serializable {
* @throws java.lang.Exception Thrown, if the state cannot be fetched.
*/
T getState() throws Exception;
+
+ /**
+ * Discards the state referred to by this handle, to free up resources in
+ * the persistent storage. This method is called when the handle will not be
+ * used any more.
+ */
+ void discardState() throws Exception;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/59bee4ad/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
index f3c40ba..8f613fc 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
@@ -53,10 +53,8 @@ import org.apache.flink.runtime.messages.TaskMessages;
import org.apache.flink.runtime.messages.TaskMessages.TaskInFinalState;
import org.apache.flink.runtime.messages.TaskManagerMessages.FatalError;
import org.apache.flink.runtime.state.StateHandle;
-
import org.apache.flink.runtime.state.StateUtils;
import org.apache.flink.runtime.util.SerializedValue;
-
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -522,7 +520,7 @@ public class Task implements Runnable {
StateUtils.setOperatorState(op, state);
}
catch (Exception e) {
- throw new Exception("Failed to deserialize state handle and setup initial operator state");
+ throw new RuntimeException("Failed to deserialize state handle and setup initial operator state.", e);
}
}
else {
http://git-wip-us.apache.org/repos/asf/flink/blob/59bee4ad/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
----------------------------------------------------------------------
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 10c8074..514b7c7 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
@@ -38,6 +38,8 @@ import java.util.List;
*/
public class CheckpointCoordinatorTest {
+ ClassLoader cl = Thread.currentThread().getContextClassLoader();
+
@Test
public void testCheckpointAbortsIfTriggerTasksAreNotExecuted() {
try {
@@ -59,7 +61,7 @@ public class CheckpointCoordinatorTest {
jid, 1, 600000,
new ExecutionVertex[] { triggerVertex1, triggerVertex2 },
new ExecutionVertex[] { ackVertex1, ackVertex2 },
- new ExecutionVertex[] {} );
+ new ExecutionVertex[] {}, cl );
// nothing should be happening
assertEquals(0, coord.getNumberOfPendingCheckpoints());
@@ -101,7 +103,7 @@ public class CheckpointCoordinatorTest {
jid, 1, 600000,
new ExecutionVertex[] { triggerVertex1, triggerVertex2 },
new ExecutionVertex[] { ackVertex1, ackVertex2 },
- new ExecutionVertex[] {} );
+ new ExecutionVertex[] {}, cl );
// nothing should be happening
assertEquals(0, coord.getNumberOfPendingCheckpoints());
@@ -139,7 +141,7 @@ public class CheckpointCoordinatorTest {
jid, 1, 600000,
new ExecutionVertex[] { vertex1, vertex2 },
new ExecutionVertex[] { vertex1, vertex2 },
- new ExecutionVertex[] { vertex1, vertex2 });
+ new ExecutionVertex[] { vertex1, vertex2 }, cl);
assertEquals(0, coord.getNumberOfPendingCheckpoints());
assertEquals(0, coord.getNumberOfRetainedSuccessfulCheckpoints());
@@ -282,7 +284,7 @@ public class CheckpointCoordinatorTest {
jid, 2, 600000,
new ExecutionVertex[] { triggerVertex1, triggerVertex2 },
new ExecutionVertex[] { ackVertex1, ackVertex2, ackVertex3 },
- new ExecutionVertex[] { commitVertex });
+ new ExecutionVertex[] { commitVertex }, cl);
assertEquals(0, coord.getNumberOfPendingCheckpoints());
assertEquals(0, coord.getNumberOfRetainedSuccessfulCheckpoints());
@@ -409,7 +411,7 @@ public class CheckpointCoordinatorTest {
jid, 10, 600000,
new ExecutionVertex[] { triggerVertex1, triggerVertex2 },
new ExecutionVertex[] { ackVertex1, ackVertex2, ackVertex3 },
- new ExecutionVertex[] { commitVertex });
+ new ExecutionVertex[] { commitVertex }, cl);
assertEquals(0, coord.getNumberOfPendingCheckpoints());
assertEquals(0, coord.getNumberOfRetainedSuccessfulCheckpoints());
@@ -523,7 +525,7 @@ public class CheckpointCoordinatorTest {
jid, 2, 200,
new ExecutionVertex[] { triggerVertex },
new ExecutionVertex[] { ackVertex1, ackVertex2 },
- new ExecutionVertex[] { commitVertex });
+ new ExecutionVertex[] { commitVertex }, cl);
// trigger a checkpoint, partially acknowledged
assertTrue(coord.triggerCheckpoint(timestamp));
@@ -581,7 +583,7 @@ public class CheckpointCoordinatorTest {
jid, 2, 200000,
new ExecutionVertex[] { triggerVertex },
new ExecutionVertex[] { ackVertex1, ackVertex2 },
- new ExecutionVertex[] { commitVertex });
+ new ExecutionVertex[] { commitVertex }, cl);
assertTrue(coord.triggerCheckpoint(timestamp));
http://git-wip-us.apache.org/repos/asf/flink/blob/59bee4ad/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointStateRestoreTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointStateRestoreTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointStateRestoreTest.java
index e1cf061..c23aaca 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointStateRestoreTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointStateRestoreTest.java
@@ -30,9 +30,7 @@ import org.apache.flink.runtime.state.LocalStateHandle;
import org.apache.flink.runtime.state.StateHandle;
import org.apache.flink.runtime.util.SerializableObject;
import org.apache.flink.runtime.util.SerializedValue;
-
import org.junit.Test;
-
import org.mockito.Mockito;
import java.util.HashMap;
@@ -47,6 +45,8 @@ import static org.mockito.Mockito.*;
*/
public class CheckpointStateRestoreTest {
+ ClassLoader cl = Thread.currentThread().getContextClassLoader();
+
@Test
public void testSetState() {
try {
@@ -82,7 +82,7 @@ public class CheckpointStateRestoreTest {
CheckpointCoordinator coord = new CheckpointCoordinator(jid, 1, 200000L,
new ExecutionVertex[] { stateful1, stateful2, stateful3, stateless1, stateless2 },
new ExecutionVertex[] { stateful1, stateful2, stateful3, stateless1, stateless2 },
- new ExecutionVertex[0]);
+ new ExecutionVertex[0], cl);
// create ourselves a checkpoint with state
final long timestamp = 34623786L;
@@ -151,7 +151,7 @@ public class CheckpointStateRestoreTest {
CheckpointCoordinator coord = new CheckpointCoordinator(jid, 1, 200000L,
new ExecutionVertex[] { stateful1, stateful2, stateful3, stateless1, stateless2 },
new ExecutionVertex[] { stateful1, stateful2, stateful3, stateless1, stateless2 },
- new ExecutionVertex[0]);
+ new ExecutionVertex[0], cl);
// create ourselves a checkpoint with state
final long timestamp = 34623786L;
@@ -191,7 +191,7 @@ public class CheckpointStateRestoreTest {
CheckpointCoordinator coord = new CheckpointCoordinator(new JobID(), 1, 200000L,
new ExecutionVertex[] { mock(ExecutionVertex.class) },
new ExecutionVertex[] { mock(ExecutionVertex.class) },
- new ExecutionVertex[0]);
+ new ExecutionVertex[0], cl);
try {
coord.restoreLatestCheckpointedState(new HashMap<JobVertexID, ExecutionJobVertex>(), true, false);
http://git-wip-us.apache.org/repos/asf/flink/blob/59bee4ad/flink-runtime/src/test/java/org/apache/flink/runtime/messages/CheckpointMessagesTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/messages/CheckpointMessagesTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/messages/CheckpointMessagesTest.java
index 9a9f486..a204c3d 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/messages/CheckpointMessagesTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/messages/CheckpointMessagesTest.java
@@ -96,5 +96,9 @@ public class CheckpointMessagesTest {
public int hashCode() {
return getClass().hashCode();
}
- };
+
+ @Override
+ public void discardState() throws Exception {
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/59bee4ad/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
index a412e05..4d4893b 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
@@ -31,6 +31,7 @@ import org.apache.flink.runtime.jobgraph.tasks.CheckpointCommittingOperator;
import org.apache.flink.runtime.jobgraph.tasks.CheckpointedOperator;
import org.apache.flink.runtime.jobgraph.tasks.OperatorStateCarrier;
import org.apache.flink.runtime.state.LocalStateHandle;
+import org.apache.flink.runtime.state.StateHandle;
import org.apache.flink.runtime.util.event.EventListener;
import org.apache.flink.streaming.api.graph.StreamConfig;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
@@ -41,7 +42,7 @@ import org.slf4j.LoggerFactory;
public abstract class StreamTask<OUT, O extends StreamOperator<OUT>> extends AbstractInvokable implements
- OperatorStateCarrier<LocalStateHandle>, CheckpointedOperator, CheckpointCommittingOperator {
+ OperatorStateCarrier<StateHandle<Serializable>>, CheckpointedOperator, CheckpointCommittingOperator {
private static final Logger LOG = LoggerFactory.getLogger(StreamTask.class);
@@ -136,7 +137,7 @@ public abstract class StreamTask<OUT, O extends StreamOperator<OUT>> extends Abs
* Re-injects the user states into the map. Also set the state on the functions.
*/
@Override
- public void setInitialState(LocalStateHandle stateHandle) throws Exception {
+ public void setInitialState(StateHandle<Serializable> stateHandle) throws Exception {
// here, we later resolve the state handle into the actual state by
// loading the state described by the handle from the backup store
Serializable state = stateHandle.getState();
@@ -184,7 +185,7 @@ public abstract class StreamTask<OUT, O extends StreamOperator<OUT>> extends Abs
LOG.info("Starting checkpoint {} on task {}", checkpointId, getName());
// first draw the state that should go into checkpoint
- LocalStateHandle state;
+ StateHandle<Serializable> state;
try {
Serializable userState = null;