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/07/26 09:30:53 UTC
[1/3] flink git commit: [FLINK-4067] [runtime] Add savepoint headers
Repository: flink
Updated Branches:
refs/heads/master 110bba386 -> f0ac261ad
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStoreITCase.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStoreITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStoreITCase.java
index 375721f..380ba2c 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStoreITCase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStoreITCase.java
@@ -57,8 +57,8 @@ public class ZooKeeperCompletedCheckpointStoreITCase extends CompletedCheckpoint
}
@Override
- protected CompletedCheckpointStore createCompletedCheckpoints(int maxNumberOfCheckpointsToRetain,
- ClassLoader userLoader) throws Exception {
+ protected CompletedCheckpointStore createCompletedCheckpoints(
+ int maxNumberOfCheckpointsToRetain, ClassLoader userLoader) throws Exception {
return new ZooKeeperCompletedCheckpointStore(maxNumberOfCheckpointsToRetain, userLoader,
ZooKeeper.createClient(), CheckpointsPath, new StateStorageHelper<CompletedCheckpoint>() {
@@ -79,7 +79,7 @@ public class ZooKeeperCompletedCheckpointStoreITCase extends CompletedCheckpoint
CompletedCheckpointStore checkpoints = createCompletedCheckpoints(3, ClassLoader
.getSystemClassLoader());
- TestCheckpoint[] expected = new TestCheckpoint[] {
+ TestCompletedCheckpoint[] expected = new TestCompletedCheckpoint[] {
createCheckpoint(0), createCheckpoint(1), createCheckpoint(2)
};
@@ -118,7 +118,7 @@ public class ZooKeeperCompletedCheckpointStoreITCase extends CompletedCheckpoint
CuratorFramework client = ZooKeeper.getClient();
CompletedCheckpointStore store = createCompletedCheckpoints(1, ClassLoader.getSystemClassLoader());
- TestCheckpoint checkpoint = createCheckpoint(0);
+ TestCompletedCheckpoint checkpoint = createCheckpoint(0);
store.addCheckpoint(checkpoint);
assertEquals(1, store.getNumberOfRetainedCheckpoints());
@@ -143,7 +143,7 @@ public class ZooKeeperCompletedCheckpointStoreITCase extends CompletedCheckpoint
CuratorFramework client = ZooKeeper.getClient();
CompletedCheckpointStore store = createCompletedCheckpoints(1, ClassLoader.getSystemClassLoader());
- TestCheckpoint checkpoint = createCheckpoint(0);
+ TestCompletedCheckpoint checkpoint = createCheckpoint(0);
store.addCheckpoint(checkpoint);
assertEquals(1, store.getNumberOfRetainedCheckpoints());
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/FsSavepointStoreTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/FsSavepointStoreTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/FsSavepointStoreTest.java
new file mode 100644
index 0000000..24ab296
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/FsSavepointStoreTest.java
@@ -0,0 +1,235 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.checkpoint.savepoint;
+
+import org.apache.flink.core.fs.FSDataOutputStream;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.checkpoint.TaskState;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.mockito.Matchers;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ThreadLocalRandom;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+
+public class FsSavepointStoreTest {
+
+ @Rule
+ public TemporaryFolder tmp = new TemporaryFolder();
+
+ /**
+ * Tests a store-load-dispose sequence.
+ */
+ @Test
+ public void testStoreLoadDispose() throws Exception {
+ FsSavepointStore store = new FsSavepointStore(tmp.getRoot().getPath(), "fs-savepoint-store-test-");
+ assertEquals(0, tmp.getRoot().listFiles().length);
+
+ // Store
+ SavepointV0 stored = new SavepointV0(1929292, SavepointV01Test.createTaskStates(4, 24));
+ String path = store.storeSavepoint(stored);
+ assertEquals(1, tmp.getRoot().listFiles().length);
+
+ // Load
+ Savepoint loaded = store.loadSavepoint(path);
+ assertEquals(stored, loaded);
+
+ // Dispose
+ store.disposeSavepoint(path, ClassLoader.getSystemClassLoader());
+
+ assertEquals(0, tmp.getRoot().listFiles().length);
+ }
+
+ /**
+ * Tests loading with unexpected magic number.
+ */
+ @Test
+ public void testUnexpectedSavepoint() throws Exception {
+ FsSavepointStore store = new FsSavepointStore(tmp.getRoot().getPath(), "fs-savepoint-store-test-");
+
+ // Random file
+ Path filePath = new Path(tmp.getRoot().getPath(), UUID.randomUUID().toString());
+ FSDataOutputStream fdos = FileSystem.get(filePath.toUri()).create(filePath, false);
+ DataOutputStream dos = new DataOutputStream(fdos);
+ for (int i = 0; i < 10; i++) {
+ dos.writeLong(ThreadLocalRandom.current().nextLong());
+ }
+
+ try {
+ store.loadSavepoint(filePath.toString());
+ fail("Did not throw expected Exception");
+ } catch (RuntimeException e) {
+ assertTrue(e.getMessage().contains("Flink 1.0") && e.getMessage().contains("Unexpected magic number"));
+ }
+ }
+
+ /**
+ * Tests addition of a new savepoint version.
+ */
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testMultipleSavepointVersions() throws Exception {
+ Field field = SavepointSerializers.class.getDeclaredField("SERIALIZERS");
+ field.setAccessible(true);
+ Map<Integer, SavepointSerializer<?>> serializers = (Map<Integer, SavepointSerializer<?>>) field.get(null);
+
+ assertTrue(serializers.size() >= 1);
+
+ FsSavepointStore store = new FsSavepointStore(tmp.getRoot().getPath(), "fs-savepoint-store-test-");
+ assertEquals(0, tmp.getRoot().listFiles().length);
+
+ // New savepoint type for test
+ int version = ThreadLocalRandom.current().nextInt();
+ long checkpointId = ThreadLocalRandom.current().nextLong();
+
+ // Add serializer
+ serializers.put(version, NewSavepointSerializer.INSTANCE);
+
+ TestSavepoint newSavepoint = new TestSavepoint(version, checkpointId);
+ String pathNewSavepoint = store.storeSavepoint(newSavepoint);
+ assertEquals(1, tmp.getRoot().listFiles().length);
+
+ // Savepoint v0
+ Savepoint savepoint = new SavepointV0(checkpointId, SavepointV01Test.createTaskStates(4, 32));
+ String pathSavepoint = store.storeSavepoint(savepoint);
+ assertEquals(2, tmp.getRoot().listFiles().length);
+
+ // Load
+ Savepoint loaded = store.loadSavepoint(pathNewSavepoint);
+ assertEquals(newSavepoint, loaded);
+
+ loaded = store.loadSavepoint(pathSavepoint);
+ assertEquals(savepoint, loaded);
+ }
+
+ /**
+ * Tests that an exception during store cleans up the created savepoint file.
+ */
+ @Test
+ public void testCleanupOnStoreFailure() throws Exception {
+ Field field = SavepointSerializers.class.getDeclaredField("SERIALIZERS");
+ field.setAccessible(true);
+ Map<Integer, SavepointSerializer<?>> serializers = (Map<Integer, SavepointSerializer<?>>) field.get(null);
+
+ final int version = 123123;
+ SavepointSerializer<TestSavepoint> serializer = mock(SavepointSerializer.class);
+ doThrow(new RuntimeException("Test Exception")).when(serializer)
+ .serialize(Matchers.any(TestSavepoint.class), any(DataOutputStream.class));
+
+ FsSavepointStore store = new FsSavepointStore(tmp.getRoot().getPath(), "fs-savepoint-store-test-");
+ serializers.put(version, serializer);
+
+ Savepoint savepoint = new TestSavepoint(version, 12123123);
+
+ assertEquals(0, tmp.getRoot().listFiles().length);
+
+ try {
+ store.storeSavepoint(savepoint);
+ } catch (Throwable ignored) {
+ }
+
+ assertEquals("Savepoint file not cleaned up on failure", 0, tmp.getRoot().listFiles().length);
+ }
+
+ private static class NewSavepointSerializer implements SavepointSerializer<TestSavepoint> {
+
+ private static final NewSavepointSerializer INSTANCE = new NewSavepointSerializer();
+
+ @Override
+ public void serialize(TestSavepoint savepoint, DataOutputStream dos) throws IOException {
+ dos.writeInt(savepoint.version);
+ dos.writeLong(savepoint.checkpointId);
+ }
+
+ @Override
+ public TestSavepoint deserialize(DataInputStream dis) throws IOException {
+ int version = dis.readInt();
+ long checkpointId = dis.readLong();
+ return new TestSavepoint(version, checkpointId);
+ }
+
+ }
+
+ private static class TestSavepoint implements Savepoint {
+
+ private final int version;
+ private final long checkpointId;
+
+ public TestSavepoint(int version, long checkpointId) {
+ this.version = version;
+ this.checkpointId = checkpointId;
+ }
+
+ @Override
+ public int getVersion() {
+ return version;
+ }
+
+ @Override
+ public long getCheckpointId() {
+ return checkpointId;
+ }
+
+ @Override
+ public Collection<TaskState> getTaskStates() {
+ return Collections.EMPTY_LIST;
+ }
+
+ @Override
+ public void dispose(ClassLoader classLoader) {
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ TestSavepoint that = (TestSavepoint) o;
+ return version == that.version && checkpointId == that.checkpointId;
+
+ }
+
+ @Override
+ public int hashCode() {
+ int result = version;
+ result = 31 * result + (int) (checkpointId ^ (checkpointId >>> 32));
+ return result;
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/HeapSavepointStoreTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/HeapSavepointStoreTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/HeapSavepointStoreTest.java
new file mode 100644
index 0000000..ec3dd0a
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/HeapSavepointStoreTest.java
@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.checkpoint.savepoint;
+
+public class HeapSavepointStoreTest {
+
+
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointCoordinatorTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointCoordinatorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointCoordinatorTest.java
new file mode 100644
index 0000000..6b4f354
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointCoordinatorTest.java
@@ -0,0 +1,1119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.checkpoint.savepoint;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.checkpoint.CheckpointIDCounter;
+import org.apache.flink.runtime.checkpoint.PendingCheckpoint;
+import org.apache.flink.runtime.checkpoint.SubtaskState;
+import org.apache.flink.runtime.checkpoint.TaskState;
+import org.apache.flink.runtime.checkpoint.stats.DisabledCheckpointStatsTracker;
+import org.apache.flink.runtime.execution.ExecutionState;
+import org.apache.flink.runtime.executiongraph.Execution;
+import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
+import org.apache.flink.runtime.executiongraph.ExecutionJobVertex;
+import org.apache.flink.runtime.executiongraph.ExecutionVertex;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.messages.checkpoint.AcknowledgeCheckpoint;
+import org.apache.flink.runtime.messages.checkpoint.DeclineCheckpoint;
+import org.apache.flink.runtime.messages.checkpoint.NotifyCheckpointComplete;
+import org.apache.flink.runtime.messages.checkpoint.TriggerCheckpoint;
+import org.apache.flink.runtime.state.LocalStateHandle;
+import org.apache.flink.runtime.state.StateHandle;
+import org.apache.flink.runtime.testutils.CommonTestUtils;
+import org.apache.flink.util.SerializedValue;
+import org.apache.flink.util.TestLogger;
+import org.junit.Test;
+import scala.concurrent.Await;
+import scala.concurrent.Future;
+import scala.concurrent.Promise;
+import scala.concurrent.duration.Deadline;
+import scala.concurrent.duration.FiniteDuration;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.Serializable;
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests for the savepoint coordinator.
+ */
+public class SavepointCoordinatorTest extends TestLogger {
+
+ // ------------------------------------------------------------------------
+ // Trigger and acknowledge
+ // ------------------------------------------------------------------------
+
+ /**
+ * Simple trigger-acknowledge test for a single savepoint.
+ */
+ @Test
+ public void testSimpleTriggerSavepoint() throws Exception {
+ JobID jobId = new JobID();
+ long checkpointTimeout = 60 * 1000;
+ long timestamp = 1272635;
+ ExecutionVertex[] vertices = new ExecutionVertex[] {
+ mockExecutionVertex(jobId),
+ mockExecutionVertex(jobId) };
+ MockCheckpointIdCounter checkpointIdCounter = new MockCheckpointIdCounter();
+ HeapSavepointStore savepointStore = new HeapSavepointStore();
+
+ SavepointCoordinator coordinator = createSavepointCoordinator(
+ jobId,
+ checkpointTimeout,
+ vertices,
+ vertices,
+ vertices,
+ checkpointIdCounter,
+ savepointStore);
+
+ // Trigger the savepoint
+ Future<String> savepointPathFuture = coordinator.triggerSavepoint(timestamp);
+ assertFalse(savepointPathFuture.isCompleted());
+
+ long checkpointId = checkpointIdCounter.getLastReturnedCount();
+ assertEquals(0, checkpointId);
+
+ // Verify send trigger messages
+ for (ExecutionVertex vertex : vertices) {
+ verifyTriggerCheckpoint(vertex, checkpointId, timestamp);
+ }
+
+ PendingCheckpoint pendingCheckpoint = coordinator.getPendingCheckpoints()
+ .get(checkpointId);
+
+ verifyPendingCheckpoint(pendingCheckpoint, jobId, checkpointId,
+ timestamp, 0, 2, 0, false, false);
+
+ // Acknowledge tasks
+ for (ExecutionVertex vertex : vertices) {
+ coordinator.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(
+ jobId, vertex.getCurrentExecutionAttempt().getAttemptId(),
+ checkpointId, createSerializedStateHandle(vertex), 0));
+ }
+
+ // The pending checkpoint is completed
+ assertTrue(pendingCheckpoint.isDiscarded());
+ assertEquals(0, coordinator.getSuccessfulCheckpoints().size());
+
+ // Verify send notify complete messages
+ for (ExecutionVertex vertex : vertices) {
+ verifyNotifyCheckpointComplete(vertex, checkpointId, timestamp);
+ }
+
+ // Verify that the future has been completed
+ assertTrue(savepointPathFuture.isCompleted());
+ String savepointPath = Await.result(savepointPathFuture, FiniteDuration.Zero());
+
+ // Verify the savepoint
+ Savepoint savepoint = savepointStore.loadSavepoint(savepointPath);
+ verifySavepoint(savepoint, checkpointId, vertices);
+
+ // Verify all promises removed
+ assertEquals(0, getSavepointPromises(coordinator).size());
+
+ coordinator.shutdown();
+ }
+
+ /**
+ * This test triggers a checkpoint and then sends a decline checkpoint message from
+ * one of the tasks. The expected behaviour is that said checkpoint is discarded and a new
+ * checkpoint is triggered.
+ */
+ @Test
+ public void testTriggerAndDeclineCheckpointSimple() throws Exception {
+ JobID jobId = new JobID();
+ long checkpointTimeout = 60 * 1000;
+ long timestamp = 1272635;
+ ExecutionVertex[] vertices = new ExecutionVertex[] {
+ mockExecutionVertex(jobId),
+ mockExecutionVertex(jobId) };
+ MockCheckpointIdCounter checkpointIdCounter = new MockCheckpointIdCounter();
+ SavepointStore savepointStore = new HeapSavepointStore();
+
+ SavepointCoordinator coordinator = createSavepointCoordinator(
+ jobId,
+ checkpointTimeout,
+ vertices,
+ vertices,
+ vertices,
+ checkpointIdCounter,
+ savepointStore);
+
+ // Trigger the savepoint
+ Future<String> savepointPathFuture = coordinator.triggerSavepoint(timestamp);
+ assertFalse(savepointPathFuture.isCompleted());
+
+ long checkpointId = checkpointIdCounter.getLastReturnedCount();
+ assertEquals(0, checkpointId);
+
+ // Verify send trigger messages
+ for (ExecutionVertex vertex : vertices) {
+ verifyTriggerCheckpoint(vertex, checkpointId, timestamp);
+ }
+
+ PendingCheckpoint pendingCheckpoint = coordinator.getPendingCheckpoints()
+ .get(checkpointId);
+
+ verifyPendingCheckpoint(pendingCheckpoint, jobId, checkpointId,
+ timestamp, 0, 2, 0, false, false);
+
+ // Acknowledge and decline tasks
+ coordinator.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(
+ jobId, vertices[0].getCurrentExecutionAttempt().getAttemptId(),
+ checkpointId, createSerializedStateHandle(vertices[0]), 0));
+
+ coordinator.receiveDeclineMessage(new DeclineCheckpoint(
+ jobId, vertices[1].getCurrentExecutionAttempt().getAttemptId(),
+ checkpointId, 0));
+
+
+ // The pending checkpoint is completed
+ assertTrue(pendingCheckpoint.isDiscarded());
+ assertEquals(0, coordinator.getSuccessfulCheckpoints().size());
+
+ // Verify that the future has been completed
+ assertTrue(savepointPathFuture.isCompleted());
+
+ try {
+ Await.result(savepointPathFuture.failed(), FiniteDuration.Zero());
+ fail("Did not throw expected exception");
+ } catch (Throwable ignored) {}
+
+ // Verify all promises removed
+ assertEquals(0, getSavepointPromises(coordinator).size());
+
+ coordinator.shutdown();
+ }
+
+ // ------------------------------------------------------------------------
+ // Rollback
+ // ------------------------------------------------------------------------
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testSimpleRollbackSavepoint() throws Exception {
+ JobID jobId = new JobID();
+
+ ExecutionJobVertex[] jobVertices = new ExecutionJobVertex[] {
+ mockExecutionJobVertex(jobId, new JobVertexID(), 4),
+ mockExecutionJobVertex(jobId, new JobVertexID(), 4) };
+
+ ExecutionVertex[] triggerVertices = jobVertices[0].getTaskVertices();
+ ExecutionVertex[] ackVertices = new ExecutionVertex[8];
+
+ int i = 0;
+ for (ExecutionJobVertex jobVertex : jobVertices) {
+ for (ExecutionVertex vertex : jobVertex.getTaskVertices()) {
+ ackVertices[i++] = vertex;
+ }
+ }
+
+ MockCheckpointIdCounter idCounter = new MockCheckpointIdCounter();
+ HeapSavepointStore savepointStore = new HeapSavepointStore();
+
+ SavepointCoordinator coordinator = createSavepointCoordinator(
+ jobId,
+ 60 * 1000,
+ triggerVertices,
+ ackVertices,
+ new ExecutionVertex[] {},
+ idCounter,
+ savepointStore);
+
+ Future<String> savepointPathFuture = coordinator.triggerSavepoint(1231273123);
+
+ // Acknowledge all tasks
+ for (ExecutionVertex vertex : ackVertices) {
+ ExecutionAttemptID attemptId = vertex.getCurrentExecutionAttempt().getAttemptId();
+ coordinator.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(
+ jobId, attemptId, 0, createSerializedStateHandle(vertex), 0));
+ }
+
+ String savepointPath = Await.result(savepointPathFuture, FiniteDuration.Zero());
+ assertNotNull(savepointPath);
+
+ // Rollback
+ coordinator.restoreSavepoint(createExecutionJobVertexMap(jobVertices), savepointPath);
+
+ // Verify all executions have been reset
+ for (ExecutionVertex vertex : ackVertices) {
+ verify(vertex.getCurrentExecutionAttempt(), times(1)).setInitialState(
+ any(SerializedValue.class), any(Map.class));
+ }
+
+ // Verify all promises removed
+ assertEquals(0, getSavepointPromises(coordinator).size());
+
+ // Verify checkpoint ID counter started
+ assertTrue(idCounter.isStarted());
+
+ coordinator.shutdown();
+ }
+
+ @Test
+ public void testRollbackParallelismMismatch() throws Exception {
+ JobID jobId = new JobID();
+
+ ExecutionJobVertex[] jobVertices = new ExecutionJobVertex[] {
+ mockExecutionJobVertex(jobId, new JobVertexID(), 4),
+ mockExecutionJobVertex(jobId, new JobVertexID(), 4) };
+
+ ExecutionVertex[] triggerVertices = jobVertices[0].getTaskVertices();
+ ExecutionVertex[] ackVertices = new ExecutionVertex[8];
+
+ int index = 0;
+ for (ExecutionJobVertex jobVertex : jobVertices) {
+ for (ExecutionVertex vertex : jobVertex.getTaskVertices()) {
+ ackVertices[index++] = vertex;
+ }
+ }
+
+ HeapSavepointStore savepointStore = new HeapSavepointStore();
+
+ SavepointCoordinator coordinator = createSavepointCoordinator(
+ jobId,
+ 60 * 1000,
+ triggerVertices,
+ ackVertices,
+ new ExecutionVertex[] {},
+ new MockCheckpointIdCounter(),
+ savepointStore);
+
+ Future<String> savepointPathFuture = coordinator.triggerSavepoint(1231273123);
+
+ // Acknowledge all tasks
+ for (ExecutionVertex vertex : ackVertices) {
+ ExecutionAttemptID attemptId = vertex.getCurrentExecutionAttempt().getAttemptId();
+ coordinator.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(
+ jobId, attemptId, 0, createSerializedStateHandle(vertex), 0));
+ }
+
+ String savepointPath = Await.result(savepointPathFuture, FiniteDuration.Zero());
+ assertNotNull(savepointPath);
+
+ // 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);
+ }
+
+ try {
+ // Rollback
+ coordinator.restoreSavepoint(
+ createExecutionJobVertexMap(jobVertices),
+ savepointPath);
+ fail("Did not throw expected Exception after rollback with parallelism mismatch.");
+ }
+ catch (Exception ignored) {
+ }
+
+ // Verify all promises removed
+ assertEquals(0, getSavepointPromises(coordinator).size());
+
+ coordinator.shutdown();
+ }
+
+ @Test
+ public void testRollbackStateStoreFailure() throws Exception {
+ JobID jobId = new JobID();
+ ExecutionJobVertex jobVertex = mockExecutionJobVertex(jobId, new JobVertexID(), 4);
+ HeapSavepointStore savepointStore = spy(new HeapSavepointStore());
+
+ SavepointCoordinator coordinator = createSavepointCoordinator(
+ jobId,
+ 60 * 1000,
+ jobVertex.getTaskVertices(),
+ jobVertex.getTaskVertices(),
+ new ExecutionVertex[] {},
+ new MockCheckpointIdCounter(),
+ savepointStore);
+
+ Future<String> savepointPathFuture = coordinator.triggerSavepoint(1231273123);
+
+ // Acknowledge all tasks
+ for (ExecutionVertex vertex : jobVertex.getTaskVertices()) {
+ ExecutionAttemptID attemptId = vertex.getCurrentExecutionAttempt().getAttemptId();
+ coordinator.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(
+ jobId, attemptId, 0, createSerializedStateHandle(vertex), 0));
+ }
+
+ String savepointPath = Await.result(savepointPathFuture, FiniteDuration.Zero());
+ assertNotNull(savepointPath);
+
+ // Failure on getState
+ doThrow(new RuntimeException("TestException")).when(savepointStore).loadSavepoint(anyString());
+
+ try {
+ // Rollback
+ coordinator.restoreSavepoint(
+ createExecutionJobVertexMap(jobVertex),
+ savepointPath);
+
+ fail("Did not throw expected Exception after rollback with savepoint store failure.");
+ }
+ catch (Exception ignored) {
+ }
+
+ // Verify all promises removed
+ assertEquals(0, getSavepointPromises(coordinator).size());
+
+ coordinator.shutdown();
+ }
+
+ @Test
+ public void testRollbackSetsCheckpointID() throws Exception {
+ SavepointV0 savepoint = new SavepointV0(12312312L, Collections.<TaskState>emptyList());
+
+ CheckpointIDCounter checkpointIdCounter = mock(CheckpointIDCounter.class);
+
+ SavepointStore savepointStore = mock(SavepointStore.class);
+ when(savepointStore.loadSavepoint(anyString())).thenReturn(savepoint);
+
+ SavepointCoordinator coordinator = createSavepointCoordinator(
+ new JobID(),
+ 60 * 1000,
+ new ExecutionVertex[] {},
+ new ExecutionVertex[] {},
+ new ExecutionVertex[] {},
+ checkpointIdCounter,
+ savepointStore);
+
+ coordinator.restoreSavepoint(createExecutionJobVertexMap(), "any");
+
+ verify(checkpointIdCounter).setCount(eq(12312312L + 1));
+
+ coordinator.shutdown();
+ }
+
+ // ------------------------------------------------------------------------
+ // Savepoint aborts and future notifications
+ // ------------------------------------------------------------------------
+
+ @Test
+ public void testAbortSavepointIfTriggerTasksNotExecuted() throws Exception {
+ JobID jobId = new JobID();
+ ExecutionVertex[] triggerVertices = new ExecutionVertex[] {
+ mock(ExecutionVertex.class),
+ mock(ExecutionVertex.class) };
+ ExecutionVertex[] ackVertices = new ExecutionVertex[] {
+ mockExecutionVertex(jobId),
+ mockExecutionVertex(jobId) };
+
+ SavepointCoordinator coordinator = createSavepointCoordinator(
+ jobId,
+ 60 * 1000,
+ triggerVertices,
+ ackVertices,
+ new ExecutionVertex[] {},
+ new MockCheckpointIdCounter(),
+ new HeapSavepointStore());
+
+ // Trigger savepoint
+ Future<String> savepointPathFuture = coordinator.triggerSavepoint(1238123);
+
+ // Abort the savepoint, because the vertices are not running
+ assertTrue(savepointPathFuture.isCompleted());
+
+ try {
+ Await.result(savepointPathFuture, FiniteDuration.Zero());
+ fail("Did not throw expected Exception after shutdown");
+ }
+ catch (Exception ignored) {
+ }
+
+ // Verify all promises removed
+ assertEquals(0, getSavepointPromises(coordinator).size());
+
+ coordinator.shutdown();
+ }
+
+ @Test
+ public void testAbortSavepointIfTriggerTasksAreFinished() throws Exception {
+ JobID jobId = new JobID();
+ ExecutionVertex[] triggerVertices = new ExecutionVertex[] {
+ mockExecutionVertex(jobId),
+ mockExecutionVertex(jobId, ExecutionState.FINISHED) };
+ ExecutionVertex[] ackVertices = new ExecutionVertex[] {
+ mockExecutionVertex(jobId),
+ mockExecutionVertex(jobId) };
+
+ SavepointCoordinator coordinator = createSavepointCoordinator(
+ jobId,
+ 60 * 1000,
+ triggerVertices,
+ ackVertices,
+ new ExecutionVertex[] {},
+ new MockCheckpointIdCounter(),
+ new HeapSavepointStore());
+
+ // Trigger savepoint
+ Future<String> savepointPathFuture = coordinator.triggerSavepoint(1238123);
+
+ // Abort the savepoint, because the vertices are not running
+ assertTrue(savepointPathFuture.isCompleted());
+
+ try {
+ Await.result(savepointPathFuture, FiniteDuration.Zero());
+ fail("Did not throw expected Exception after shutdown");
+ }
+ catch (Exception ignored) {
+ }
+
+ // Verify all promises removed
+ assertEquals(0, getSavepointPromises(coordinator).size());
+
+ coordinator.shutdown();
+ }
+
+ @Test
+ public void testAbortSavepointIfAckTasksAreNotExecuted() throws Exception {
+ JobID jobId = new JobID();
+ ExecutionVertex[] triggerVertices = new ExecutionVertex[] {
+ mockExecutionVertex(jobId),
+ mockExecutionVertex(jobId) };
+ ExecutionVertex[] ackVertices = new ExecutionVertex[] {
+ mock(ExecutionVertex.class),
+ mock(ExecutionVertex.class) };
+
+ SavepointCoordinator coordinator = createSavepointCoordinator(
+ jobId,
+ 60 * 1000,
+ triggerVertices,
+ ackVertices,
+ new ExecutionVertex[] {},
+ new MockCheckpointIdCounter(),
+ new HeapSavepointStore());
+
+ // Trigger savepoint
+ Future<String> savepointPathFuture = coordinator.triggerSavepoint(1238123);
+
+ // Abort the savepoint, because the vertices are not running
+ assertTrue(savepointPathFuture.isCompleted());
+
+ try {
+ Await.result(savepointPathFuture, FiniteDuration.Zero());
+ fail("Did not throw expected Exception after shutdown");
+ }
+ catch (Exception ignored) {
+ }
+
+ // Verify all promises removed
+ assertEquals(0, getSavepointPromises(coordinator).size());
+
+ coordinator.shutdown();
+ }
+
+ @Test
+ public void testAbortOnCheckpointTimeout() throws Exception {
+ JobID jobId = new JobID();
+ ExecutionVertex[] vertices = new ExecutionVertex[] {
+ mockExecutionVertex(jobId),
+ mockExecutionVertex(jobId) };
+ ExecutionVertex commitVertex = mockExecutionVertex(jobId);
+ MockCheckpointIdCounter checkpointIdCounter = new MockCheckpointIdCounter();
+
+ long checkpointTimeout = 1000;
+ SavepointCoordinator coordinator = createSavepointCoordinator(
+ jobId,
+ checkpointTimeout,
+ vertices,
+ vertices,
+ new ExecutionVertex[] { commitVertex },
+ checkpointIdCounter,
+ new HeapSavepointStore());
+
+ // Trigger the savepoint
+ Future<String> savepointPathFuture = coordinator.triggerSavepoint(12731273);
+ assertFalse(savepointPathFuture.isCompleted());
+
+ long checkpointId = checkpointIdCounter.getLastReturnedCount();
+ PendingCheckpoint pendingCheckpoint = coordinator.getPendingCheckpoints()
+ .get(checkpointId);
+
+ assertNotNull("Checkpoint not pending (test race)", pendingCheckpoint);
+ assertFalse("Checkpoint already discarded (test race)", pendingCheckpoint.isDiscarded());
+
+ // Wait for savepoint to timeout
+ Deadline deadline = FiniteDuration.apply(60, "s").fromNow();
+ while (deadline.hasTimeLeft()
+ && !pendingCheckpoint.isDiscarded()
+ && coordinator.getNumberOfPendingCheckpoints() > 0) {
+
+ Thread.sleep(250);
+ }
+
+ // Verify discarded
+ assertTrue("Savepoint not discarded within timeout", pendingCheckpoint.isDiscarded());
+ assertEquals(0, coordinator.getNumberOfPendingCheckpoints());
+ assertEquals(0, coordinator.getNumberOfRetainedSuccessfulCheckpoints());
+
+ // No commit for timeout
+ verify(commitVertex, times(0)).sendMessageToCurrentExecution(
+ any(NotifyCheckpointComplete.class), any(ExecutionAttemptID.class));
+
+ assertTrue(savepointPathFuture.isCompleted());
+
+ try {
+ Await.result(savepointPathFuture, FiniteDuration.Zero());
+ fail("Did not throw expected Exception after timeout");
+ }
+ catch (Exception ignored) {
+ }
+
+ // Verify all promises removed
+ assertEquals(0, getSavepointPromises(coordinator).size());
+
+ coordinator.shutdown();
+ }
+
+ @Test
+ public void testAbortSavepointsOnShutdown() throws Exception {
+ JobID jobId = new JobID();
+ ExecutionVertex[] vertices = new ExecutionVertex[] {
+ mockExecutionVertex(jobId),
+ mockExecutionVertex(jobId) };
+
+ SavepointCoordinator coordinator = createSavepointCoordinator(
+ jobId,
+ 60 * 1000,
+ vertices,
+ vertices,
+ vertices,
+ new MockCheckpointIdCounter(),
+ new HeapSavepointStore());
+
+ // Trigger savepoints
+ List<Future<String>> savepointPathFutures = new ArrayList<>();
+ savepointPathFutures.add(coordinator.triggerSavepoint(12731273));
+ savepointPathFutures.add(coordinator.triggerSavepoint(12731273 + 123));
+
+ for (Future<String> future : savepointPathFutures) {
+ assertFalse(future.isCompleted());
+ }
+
+ coordinator.shutdown();
+
+ // Verify futures failed
+ for (Future<String> future : savepointPathFutures) {
+ assertTrue(future.isCompleted());
+
+ try {
+ Await.result(future, FiniteDuration.Zero());
+ fail("Did not throw expected Exception after shutdown");
+ }
+ catch (Exception ignored) {
+ }
+ }
+
+ // Verify all promises removed
+ assertEquals(0, getSavepointPromises(coordinator).size());
+ }
+
+ @Test
+ public void testAbortSavepointOnStateStoreFailure() throws Exception {
+ JobID jobId = new JobID();
+ ExecutionJobVertex jobVertex = mockExecutionJobVertex(jobId, new JobVertexID(), 4);
+ HeapSavepointStore savepointStore = spy(new HeapSavepointStore());
+
+ SavepointCoordinator coordinator = createSavepointCoordinator(
+ jobId,
+ 60 * 1000,
+ jobVertex.getTaskVertices(),
+ jobVertex.getTaskVertices(),
+ new ExecutionVertex[] {},
+ new MockCheckpointIdCounter(),
+ savepointStore);
+
+ // Failure on putState
+ doThrow(new RuntimeException("TestException"))
+ .when(savepointStore).storeSavepoint(any(Savepoint.class));
+
+ Future<String> savepointPathFuture = coordinator.triggerSavepoint(1231273123);
+
+ // Acknowledge all tasks
+ for (ExecutionVertex vertex : jobVertex.getTaskVertices()) {
+ ExecutionAttemptID attemptId = vertex.getCurrentExecutionAttempt().getAttemptId();
+ coordinator.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(
+ jobId, attemptId, 0, createSerializedStateHandle(vertex), 0));
+ }
+
+ try {
+ Await.result(savepointPathFuture, FiniteDuration.Zero());
+ fail("Did not throw expected Exception after rollback with savepoint store failure.");
+ }
+ catch (Exception ignored) {
+ }
+
+ // Verify all promises removed
+ assertEquals(0, getSavepointPromises(coordinator).size());
+
+ coordinator.shutdown();
+ }
+
+ @Test
+ public void testAbortSavepointIfSubsumed() throws Exception {
+ JobID jobId = new JobID();
+ long checkpointTimeout = 60 * 1000;
+ long[] timestamps = new long[] { 1272635, 1272635 + 10 };
+ long[] checkpointIds = new long[2];
+ ExecutionVertex[] vertices = new ExecutionVertex[] {
+ mockExecutionVertex(jobId),
+ mockExecutionVertex(jobId) };
+ MockCheckpointIdCounter checkpointIdCounter = new MockCheckpointIdCounter();
+ HeapSavepointStore savepointStore = new HeapSavepointStore();
+
+ SavepointCoordinator coordinator = createSavepointCoordinator(
+ jobId,
+ checkpointTimeout,
+ vertices,
+ vertices,
+ vertices,
+ checkpointIdCounter,
+ savepointStore);
+
+ // Trigger the savepoints
+ List<Future<String>> savepointPathFutures = new ArrayList<>();
+
+ savepointPathFutures.add(coordinator.triggerSavepoint(timestamps[0]));
+ checkpointIds[0] = checkpointIdCounter.getLastReturnedCount();
+
+ savepointPathFutures.add(coordinator.triggerSavepoint(timestamps[1]));
+ checkpointIds[1] = checkpointIdCounter.getLastReturnedCount();
+
+ for (Future<String> future : savepointPathFutures) {
+ assertFalse(future.isCompleted());
+ }
+
+ // Verify send trigger messages
+ for (ExecutionVertex vertex : vertices) {
+ verifyTriggerCheckpoint(vertex, checkpointIds[0], timestamps[0]);
+ verifyTriggerCheckpoint(vertex, checkpointIds[1], timestamps[1]);
+ }
+
+ PendingCheckpoint[] pendingCheckpoints = new PendingCheckpoint[] {
+ coordinator.getPendingCheckpoints().get(checkpointIds[0]),
+ coordinator.getPendingCheckpoints().get(checkpointIds[1]) };
+
+ verifyPendingCheckpoint(pendingCheckpoints[0], jobId, checkpointIds[0],
+ timestamps[0], 0, 2, 0, false, false);
+
+ verifyPendingCheckpoint(pendingCheckpoints[1], jobId, checkpointIds[1],
+ timestamps[1], 0, 2, 0, false, false);
+
+ // Acknowledge second checkpoint...
+ for (ExecutionVertex vertex : vertices) {
+ coordinator.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(
+ jobId, vertex.getCurrentExecutionAttempt().getAttemptId(),
+ checkpointIds[1], createSerializedStateHandle(vertex), 0));
+ }
+
+ // ...and one task of first checkpoint
+ coordinator.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(
+ jobId, vertices[0].getCurrentExecutionAttempt().getAttemptId(),
+ checkpointIds[0], createSerializedStateHandle(vertices[0]), 0));
+
+ // The second pending checkpoint is completed and subsumes the first one
+ assertTrue(pendingCheckpoints[0].isDiscarded());
+ assertTrue(pendingCheckpoints[1].isDiscarded());
+ assertEquals(0, coordinator.getSuccessfulCheckpoints().size());
+
+ // Verify send notify complete messages for second checkpoint
+ for (ExecutionVertex vertex : vertices) {
+ verifyNotifyCheckpointComplete(vertex, checkpointIds[1], timestamps[1]);
+ }
+
+ Savepoint[] savepoints = new Savepoint[2];
+ String[] savepointPaths = new String[2];
+
+ // Verify that the futures have both been completed
+ assertTrue(savepointPathFutures.get(0).isCompleted());
+
+ try {
+ savepointPaths[0] = Await.result(savepointPathFutures.get(0), FiniteDuration.Zero());
+ fail("Did not throw expected exception");
+ }
+ catch (Exception ignored) {
+ }
+
+ // Verify the second savepoint
+ assertTrue(savepointPathFutures.get(1).isCompleted());
+ savepointPaths[1] = Await.result(savepointPathFutures.get(1), FiniteDuration.Zero());
+ savepoints[1] = savepointStore.loadSavepoint(savepointPaths[1]);
+ verifySavepoint(savepoints[1], checkpointIds[1], vertices);
+
+ // Verify all promises removed
+ assertEquals(0, getSavepointPromises(coordinator).size());
+
+ coordinator.shutdown();
+ }
+
+ @Test
+ public void testShutdownDoesNotCleanUpCompletedCheckpointsWithFileSystemStore() throws Exception {
+ JobID jobId = new JobID();
+ long checkpointTimeout = 60 * 1000;
+ long timestamp = 1272635;
+ ExecutionVertex[] vertices = new ExecutionVertex[] {
+ mockExecutionVertex(jobId),
+ mockExecutionVertex(jobId) };
+ MockCheckpointIdCounter checkpointIdCounter = new MockCheckpointIdCounter();
+
+ // Temporary directory for file state backend
+ final File tmpDir = CommonTestUtils.createTempDirectory();
+
+ try {
+ FsSavepointStore savepointStore = new FsSavepointStore(tmpDir.toURI().toString(), "sp-");
+
+ SavepointCoordinator coordinator = createSavepointCoordinator(
+ jobId,
+ checkpointTimeout,
+ vertices,
+ vertices,
+ vertices,
+ checkpointIdCounter,
+ savepointStore);
+
+ // Trigger the savepoint
+ Future<String> savepointPathFuture = coordinator.triggerSavepoint(timestamp);
+ assertFalse(savepointPathFuture.isCompleted());
+
+ long checkpointId = checkpointIdCounter.getLastReturnedCount();
+ assertEquals(0, checkpointId);
+
+ // Verify send trigger messages
+ for (ExecutionVertex vertex : vertices) {
+ verifyTriggerCheckpoint(vertex, checkpointId, timestamp);
+ }
+
+ PendingCheckpoint pendingCheckpoint = coordinator.getPendingCheckpoints()
+ .get(checkpointId);
+
+ verifyPendingCheckpoint(pendingCheckpoint, jobId, checkpointId,
+ timestamp, 0, 2, 0, false, false);
+
+ // Acknowledge tasks
+ for (ExecutionVertex vertex : vertices) {
+ coordinator.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(
+ jobId, vertex.getCurrentExecutionAttempt().getAttemptId(),
+ checkpointId, createSerializedStateHandle(vertex), 0));
+ }
+
+ // The pending checkpoint is completed
+ assertTrue(pendingCheckpoint.isDiscarded());
+ assertEquals(0, coordinator.getSuccessfulCheckpoints().size());
+
+ // Verify send notify complete messages
+ for (ExecutionVertex vertex : vertices) {
+ verifyNotifyCheckpointComplete(vertex, checkpointId, timestamp);
+ }
+
+ // Verify that the future has been completed
+ assertTrue(savepointPathFuture.isCompleted());
+ String savepointPath = Await.result(savepointPathFuture, FiniteDuration.Zero());
+
+ // Verify all promises removed
+ assertEquals(0, getSavepointPromises(coordinator).size());
+
+ coordinator.shutdown();
+
+ // Verify the savepoint is still available
+ Savepoint savepoint = savepointStore.loadSavepoint(savepointPath);
+ verifySavepoint(savepoint, checkpointId, vertices);
+ }
+ finally {
+ FileUtils.deleteDirectory(tmpDir);
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // Test helpers
+ // ------------------------------------------------------------------------
+
+ private static SavepointCoordinator createSavepointCoordinator(
+ JobID jobId,
+ long checkpointTimeout,
+ ExecutionVertex[] triggerVertices,
+ ExecutionVertex[] ackVertices,
+ ExecutionVertex[] commitVertices,
+ CheckpointIDCounter checkpointIdCounter,
+ SavepointStore savepointStore) throws Exception {
+
+ ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
+
+ return new SavepointCoordinator(
+ jobId,
+ checkpointTimeout,
+ checkpointTimeout,
+ 42,
+ triggerVertices,
+ ackVertices,
+ commitVertices,
+ classLoader,
+ checkpointIdCounter,
+ savepointStore,
+ new DisabledCheckpointStatsTracker());
+ }
+
+ private static Map<JobVertexID, ExecutionJobVertex> createExecutionJobVertexMap(
+ ExecutionJobVertex... jobVertices) {
+
+ Map<JobVertexID, ExecutionJobVertex> jobVertexMap = new HashMap<>();
+
+ for (ExecutionJobVertex jobVertex : jobVertices) {
+ jobVertexMap.put(jobVertex.getJobVertexId(), jobVertex);
+ }
+
+ return jobVertexMap;
+ }
+
+ private static SerializedValue<StateHandle<?>> createSerializedStateHandle(
+ ExecutionVertex vertex) throws IOException {
+
+ return new SerializedValue<StateHandle<?>>(new LocalStateHandle<Serializable>(
+ vertex.getCurrentExecutionAttempt().getAttemptId()));
+ }
+
+ @SuppressWarnings("unchecked")
+ private Map<Long, Promise<String>> getSavepointPromises(
+ SavepointCoordinator coordinator)
+ throws NoSuchFieldException, IllegalAccessException {
+
+ Field field = SavepointCoordinator.class.getDeclaredField("savepointPromises");
+ field.setAccessible(true);
+ return (Map<Long, Promise<String>>) field.get(coordinator);
+ }
+
+ // ---- Verification ------------------------------------------------------
+
+ private static void verifyTriggerCheckpoint(
+ ExecutionVertex mockExecutionVertex,
+ long expectedCheckpointId,
+ long expectedTimestamp) {
+
+ ExecutionAttemptID attemptId = mockExecutionVertex
+ .getCurrentExecutionAttempt().getAttemptId();
+
+ TriggerCheckpoint expectedMsg = new TriggerCheckpoint(
+ mockExecutionVertex.getJobId(),
+ attemptId,
+ expectedCheckpointId,
+ expectedTimestamp);
+
+ verify(mockExecutionVertex).sendMessageToCurrentExecution(
+ eq(expectedMsg), eq(attemptId));
+ }
+
+ private static void verifyNotifyCheckpointComplete(
+ ExecutionVertex mockExecutionVertex,
+ long expectedCheckpointId,
+ long expectedTimestamp) {
+
+ ExecutionAttemptID attemptId = mockExecutionVertex
+ .getCurrentExecutionAttempt().getAttemptId();
+
+ NotifyCheckpointComplete expectedMsg = new NotifyCheckpointComplete(
+ mockExecutionVertex.getJobId(),
+ attemptId,
+ expectedCheckpointId,
+ expectedTimestamp);
+
+ verify(mockExecutionVertex).sendMessageToCurrentExecution(
+ eq(expectedMsg), eq(attemptId));
+ }
+
+ private static void verifyPendingCheckpoint(
+ PendingCheckpoint checkpoint,
+ JobID expectedJobId,
+ long expectedCheckpointId,
+ long expectedTimestamp,
+ int expectedNumberOfAcknowledgedTasks,
+ int expectedNumberOfNonAcknowledgedTasks,
+ int expectedNumberOfCollectedStates,
+ boolean expectedIsDiscarded,
+ boolean expectedIsFullyAcknowledged) {
+
+ assertNotNull(checkpoint);
+ assertEquals(expectedJobId, checkpoint.getJobId());
+ assertEquals(expectedCheckpointId, checkpoint.getCheckpointId());
+ assertEquals(expectedTimestamp, checkpoint.getCheckpointTimestamp());
+ assertEquals(expectedNumberOfAcknowledgedTasks, checkpoint.getNumberOfAcknowledgedTasks());
+ assertEquals(expectedNumberOfNonAcknowledgedTasks, checkpoint.getNumberOfNonAcknowledgedTasks());
+
+ int actualNumberOfCollectedStates = 0;
+
+ for (TaskState taskState : checkpoint.getTaskStates().values()) {
+ actualNumberOfCollectedStates += taskState.getNumberCollectedStates();
+ }
+
+ assertEquals(expectedNumberOfCollectedStates, actualNumberOfCollectedStates);
+ assertEquals(expectedIsDiscarded, checkpoint.isDiscarded());
+ assertEquals(expectedIsFullyAcknowledged, checkpoint.isFullyAcknowledged());
+ }
+
+ private static void verifySavepoint(
+ Savepoint savepoint,
+ long expectedCheckpointId,
+ ExecutionVertex[] expectedVertices) throws Exception {
+
+ assertEquals(expectedCheckpointId, savepoint.getCheckpointId());
+
+ for (TaskState taskState : savepoint.getTaskStates()) {
+ JobVertexID jobVertexId = taskState.getJobVertexID();
+
+ // Find matching execution vertex
+ ExecutionVertex vertex = null;
+ for (ExecutionVertex executionVertex : expectedVertices) {
+ if (executionVertex.getJobvertexId().equals(jobVertexId)) {
+ vertex = executionVertex;
+ break;
+ }
+ }
+
+ if (vertex == null) {
+ fail("Did not find matching vertex");
+ } else {
+ SubtaskState subtaskState = taskState.getState(vertex.getParallelSubtaskIndex());
+ ExecutionAttemptID vertexAttemptId = vertex.getCurrentExecutionAttempt().getAttemptId();
+
+ ExecutionAttemptID stateAttemptId = (ExecutionAttemptID) subtaskState.getState()
+ .deserializeValue(Thread.currentThread().getContextClassLoader())
+ .getState(Thread.currentThread().getContextClassLoader());
+
+ assertEquals(vertexAttemptId, stateAttemptId);
+ }
+ }
+ }
+
+ // ---- Mocking -----------------------------------------------------------
+
+ private static ExecutionJobVertex mockExecutionJobVertex(
+ JobID jobId,
+ JobVertexID jobVertexId,
+ int parallelism) {
+
+ ExecutionJobVertex jobVertex = mock(ExecutionJobVertex.class);
+ when(jobVertex.getJobId()).thenReturn(jobId);
+ when(jobVertex.getJobVertexId()).thenReturn(jobVertexId);
+ when(jobVertex.getParallelism()).thenReturn(parallelism);
+
+ ExecutionVertex[] vertices = new ExecutionVertex[parallelism];
+
+ for (int i = 0; i < vertices.length; i++) {
+ vertices[i] = mockExecutionVertex(jobId, jobVertexId, i, parallelism, ExecutionState.RUNNING);
+ }
+
+ when(jobVertex.getTaskVertices()).thenReturn(vertices);
+
+ return jobVertex;
+ }
+
+ private static ExecutionVertex mockExecutionVertex(JobID jobId) {
+ return mockExecutionVertex(jobId, ExecutionState.RUNNING);
+ }
+
+ private static ExecutionVertex mockExecutionVertex(
+ JobID jobId,
+ ExecutionState state) {
+
+ return mockExecutionVertex(jobId, new JobVertexID(), 0, 1, state);
+ }
+
+ private static ExecutionVertex mockExecutionVertex(
+ JobID jobId,
+ JobVertexID jobVertexId,
+ int subtaskIndex,
+ int parallelism,
+ ExecutionState executionState) {
+
+ Execution exec = mock(Execution.class);
+ when(exec.getAttemptId()).thenReturn(new ExecutionAttemptID());
+ when(exec.getState()).thenReturn(executionState);
+
+ ExecutionVertex vertex = mock(ExecutionVertex.class);
+ when(vertex.getJobId()).thenReturn(jobId);
+ when(vertex.getJobvertexId()).thenReturn(jobVertexId);
+ when(vertex.getParallelSubtaskIndex()).thenReturn(subtaskIndex);
+ when(vertex.getCurrentExecutionAttempt()).thenReturn(exec);
+ when(vertex.getTotalNumberOfParallelSubtasks()).thenReturn(parallelism);
+
+ return vertex;
+ }
+
+ private static class MockCheckpointIdCounter implements CheckpointIDCounter {
+
+ private boolean started;
+ private long count;
+ private long lastReturnedCount;
+
+ @Override
+ public void start() throws Exception {
+ started = true;
+ }
+
+ @Override
+ public void shutdown() throws Exception {
+ started = false;
+ }
+
+ @Override
+ public void suspend() throws Exception {
+ started = false;
+ }
+
+ @Override
+ public long getAndIncrement() throws Exception {
+ lastReturnedCount = count;
+ return count++;
+ }
+
+ @Override
+ public void setCount(long newCount) {
+ count = newCount;
+ }
+
+ long getLastReturnedCount() {
+ return lastReturnedCount;
+ }
+
+ public boolean isStarted() {
+ return started;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointStoreFactoryTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointStoreFactoryTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointStoreFactoryTest.java
new file mode 100644
index 0000000..3dfe85e
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointStoreFactoryTest.java
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.checkpoint.savepoint;
+
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.IllegalConfigurationException;
+import org.apache.flink.core.fs.Path;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class SavepointStoreFactoryTest {
+
+ @Test
+ public void testStateStoreWithDefaultConfig() throws Exception {
+ SavepointStore store = SavepointStoreFactory.createFromConfig(new Configuration());
+ Assert.assertTrue(store instanceof HeapSavepointStore);
+ }
+
+ @Test
+ public void testSavepointBackendJobManager() throws Exception {
+ Configuration config = new Configuration();
+ config.setString(SavepointStoreFactory.SAVEPOINT_BACKEND_KEY, "jobmanager");
+ SavepointStore store = SavepointStoreFactory.createFromConfig(config);
+ Assert.assertTrue(store instanceof HeapSavepointStore);
+ }
+
+ @Test
+ public void testSavepointBackendFileSystem() throws Exception {
+ Configuration config = new Configuration();
+ String rootPath = System.getProperty("java.io.tmpdir");
+ config.setString(ConfigConstants.STATE_BACKEND, "filesystem");
+ config.setString(SavepointStoreFactory.SAVEPOINT_BACKEND_KEY, "filesystem");
+ config.setString(SavepointStoreFactory.SAVEPOINT_DIRECTORY_KEY, rootPath);
+
+ SavepointStore store = SavepointStoreFactory.createFromConfig(config);
+ Assert.assertTrue(store instanceof FsSavepointStore);
+
+ FsSavepointStore stateStore = (FsSavepointStore) store;
+ Assert.assertEquals(new Path(rootPath), stateStore.getRootPath());
+ }
+
+ @Test
+ public void testSavepointBackendFileSystemButCheckpointBackendJobManager() throws Exception {
+ Configuration config = new Configuration();
+ String rootPath = System.getProperty("java.io.tmpdir");
+ // This combination does not make sense, because the checkpoints will be
+ // lost after the job manager shuts down.
+ config.setString(ConfigConstants.STATE_BACKEND, "jobmanager");
+ config.setString(SavepointStoreFactory.SAVEPOINT_BACKEND_KEY, "filesystem");
+ config.setString(SavepointStoreFactory.SAVEPOINT_DIRECTORY_KEY, rootPath);
+
+ SavepointStore store = SavepointStoreFactory.createFromConfig(config);
+ Assert.assertTrue(store instanceof FsSavepointStore);
+
+ FsSavepointStore stateStore = (FsSavepointStore) store;
+ Assert.assertEquals(new Path(rootPath), stateStore.getRootPath());
+ }
+
+ @Test(expected = IllegalConfigurationException.class)
+ public void testSavepointBackendFileSystemButNoDirectory() throws Exception {
+ Configuration config = new Configuration();
+ config.setString(SavepointStoreFactory.SAVEPOINT_BACKEND_KEY, "filesystem");
+ SavepointStoreFactory.createFromConfig(config);
+ Assert.fail("Did not throw expected Exception");
+ }
+
+ @Test(expected = IllegalConfigurationException.class)
+ public void testUnexpectedSavepointBackend() throws Exception {
+ Configuration config = new Configuration();
+ config.setString(SavepointStoreFactory.SAVEPOINT_BACKEND_KEY, "unexpected");
+ SavepointStoreFactory.createFromConfig(config);
+ Assert.fail("Did not throw expected Exception");
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointV01SerializerTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointV01SerializerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointV01SerializerTest.java
new file mode 100644
index 0000000..613bb61
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointV01SerializerTest.java
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.checkpoint.savepoint;
+
+import org.apache.commons.io.output.ByteArrayOutputStream;
+import org.apache.flink.core.memory.DataInputViewStreamWrapper;
+import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
+import org.junit.Test;
+
+import java.io.ByteArrayInputStream;
+
+import static org.junit.Assert.assertEquals;
+
+public class SavepointV01SerializerTest {
+
+ /**
+ * Test serialization of {@link SavepointV0} instance.
+ */
+ @Test
+ public void testSerializeDeserializeV1() throws Exception {
+ SavepointV0 expected = new SavepointV0(123123, SavepointV01Test.createTaskStates(8, 32));
+
+ SavepointV1Serializer serializer = SavepointV1Serializer.INSTANCE;
+
+ // Serialize
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ serializer.serialize(expected, new DataOutputViewStreamWrapper(baos));
+ byte[] bytes = baos.toByteArray();
+
+ // Deserialize
+ ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
+ Savepoint actual = serializer.deserialize(new DataInputViewStreamWrapper(bais));
+
+ assertEquals(expected, actual);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointV01Test.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointV01Test.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointV01Test.java
new file mode 100644
index 0000000..1a66afb
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointV01Test.java
@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.checkpoint.savepoint;
+
+import org.apache.flink.runtime.checkpoint.SubtaskState;
+import org.apache.flink.runtime.checkpoint.TaskState;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.messages.CheckpointMessagesTest;
+import org.apache.flink.runtime.state.StateHandle;
+import org.apache.flink.util.SerializedValue;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.ThreadLocalRandom;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class SavepointV01Test {
+
+ /**
+ * Simple test of savepoint methods.
+ */
+ @Test
+ public void testSavepointV1() throws Exception {
+ long checkpointId = ThreadLocalRandom.current().nextLong(Integer.MAX_VALUE);
+ int numTaskStates = 4;
+ int numSubtaskStates = 16;
+
+ Collection<TaskState> expected = createTaskStates(numTaskStates, numSubtaskStates);
+
+ SavepointV0 savepoint = new SavepointV0(checkpointId, expected);
+
+ assertEquals(SavepointV0.VERSION, savepoint.getVersion());
+ assertEquals(checkpointId, savepoint.getCheckpointId());
+ assertEquals(expected, savepoint.getTaskStates());
+
+ assertFalse(savepoint.getTaskStates().isEmpty());
+ savepoint.dispose(ClassLoader.getSystemClassLoader());
+ assertTrue(savepoint.getTaskStates().isEmpty());
+ }
+
+ static Collection<TaskState> createTaskStates(int numTaskStates, int numSubtaskStates) throws IOException {
+ List<TaskState> taskStates = new ArrayList<>(numTaskStates);
+
+ for (int i = 0; i < numTaskStates; i++) {
+ TaskState taskState = new TaskState(new JobVertexID(), numSubtaskStates);
+ for (int j = 0; j < numSubtaskStates; j++) {
+ SerializedValue<StateHandle<?>> stateHandle = new SerializedValue<StateHandle<?>>(
+ new CheckpointMessagesTest.MyHandle());
+
+ taskState.putState(i, new SubtaskState(stateHandle, 0, 0));
+ }
+
+ taskStates.add(taskState);
+ }
+
+ return taskStates;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobManagerHARecoveryTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobManagerHARecoveryTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobManagerHARecoveryTest.java
index a7492ea..f050e29 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobManagerHARecoveryTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobManagerHARecoveryTest.java
@@ -33,9 +33,8 @@ import org.apache.flink.runtime.checkpoint.CheckpointIDCounter;
import org.apache.flink.runtime.checkpoint.CheckpointRecoveryFactory;
import org.apache.flink.runtime.checkpoint.CompletedCheckpoint;
import org.apache.flink.runtime.checkpoint.CompletedCheckpointStore;
-import org.apache.flink.runtime.checkpoint.HeapStateStore;
-import org.apache.flink.runtime.checkpoint.SavepointStore;
import org.apache.flink.runtime.checkpoint.StandaloneCheckpointIDCounter;
+import org.apache.flink.runtime.checkpoint.savepoint.HeapSavepointStore;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.execution.librarycache.BlobLibraryCacheManager;
import org.apache.flink.runtime.executiongraph.restart.FixedDelayRestartStrategy;
@@ -158,7 +157,7 @@ public class JobManagerHARecoveryTest {
myLeaderElectionService,
mySubmittedJobGraphStore,
checkpointStateFactory,
- new SavepointStore(new HeapStateStore()),
+ new HeapSavepointStore(),
jobRecoveryTimeout,
Option.apply(null));
@@ -349,7 +348,7 @@ public class JobManagerHARecoveryTest {
public void stop() {}
@Override
- public CompletedCheckpointStore createCompletedCheckpoints(JobID jobId, ClassLoader userClassLoader) throws Exception {
+ public CompletedCheckpointStore createCheckpointStore(JobID jobId, ClassLoader userClassLoader) throws Exception {
return store;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/JobManagerLeaderElectionTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/JobManagerLeaderElectionTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/JobManagerLeaderElectionTest.java
index 6b64d6f..bdd019d 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/JobManagerLeaderElectionTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/JobManagerLeaderElectionTest.java
@@ -31,8 +31,6 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.blob.BlobServer;
import org.apache.flink.runtime.checkpoint.CheckpointRecoveryFactory;
-import org.apache.flink.runtime.checkpoint.SavepointStore;
-import org.apache.flink.runtime.checkpoint.SavepointStoreFactory;
import org.apache.flink.runtime.checkpoint.StandaloneCheckpointRecoveryFactory;
import org.apache.flink.runtime.execution.librarycache.BlobLibraryCacheManager;
import org.apache.flink.runtime.executiongraph.restart.NoRestartStrategy;
@@ -41,6 +39,8 @@ import org.apache.flink.runtime.jobmanager.RecoveryMode;
import org.apache.flink.runtime.jobmanager.StandaloneSubmittedJobGraphStore;
import org.apache.flink.runtime.jobmanager.SubmittedJobGraphStore;
import org.apache.flink.runtime.jobmanager.scheduler.Scheduler;
+import org.apache.flink.runtime.checkpoint.savepoint.SavepointStore;
+import org.apache.flink.runtime.checkpoint.savepoint.SavepointStoreFactory;
import org.apache.flink.runtime.testingUtils.TestingJobManager;
import org.apache.flink.runtime.testingUtils.TestingJobManagerMessages;
import org.apache.flink.runtime.testingUtils.TestingUtils;
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/JobManagerITCase.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/JobManagerITCase.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/JobManagerITCase.scala
index 2b5b29f..0c5534a 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/JobManagerITCase.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/JobManagerITCase.scala
@@ -24,13 +24,14 @@ import akka.testkit.{ImplicitSender, TestKit}
import akka.util.Timeout
import org.apache.flink.api.common.{ExecutionConfig, JobID}
import org.apache.flink.runtime.akka.ListeningBehaviour
-import org.apache.flink.runtime.checkpoint.{CheckpointCoordinator, SavepointCoordinator}
+import org.apache.flink.runtime.checkpoint.CheckpointCoordinator
import org.apache.flink.runtime.client.JobExecutionException
import org.apache.flink.runtime.jobgraph.tasks.JobSnapshottingSettings
import org.apache.flink.runtime.jobgraph.{DistributionPattern, JobGraph, JobVertex, ScheduleMode}
import org.apache.flink.runtime.jobmanager.Tasks._
import org.apache.flink.runtime.jobmanager.scheduler.{NoResourceAvailableException, SlotSharingGroup}
import org.apache.flink.runtime.messages.JobManagerMessages._
+import org.apache.flink.runtime.checkpoint.savepoint.SavepointCoordinator
import org.apache.flink.runtime.testingUtils.TestingJobManagerMessages._
import org.apache.flink.runtime.testingUtils.{ScalaTestingUtils, TestingUtils}
import org.apache.flink.runtime.testutils.JobManagerActorTestUtils
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/SavepointITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/SavepointITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/SavepointITCase.java
index c4e0f42..0ed28ad 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/SavepointITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/SavepointITCase.java
@@ -31,10 +31,10 @@ import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.blob.BlobKey;
-import org.apache.flink.runtime.checkpoint.CompletedCheckpoint;
-import org.apache.flink.runtime.checkpoint.SavepointStoreFactory;
import org.apache.flink.runtime.checkpoint.SubtaskState;
import org.apache.flink.runtime.checkpoint.TaskState;
+import org.apache.flink.runtime.checkpoint.savepoint.SavepointStoreFactory;
+import org.apache.flink.runtime.checkpoint.savepoint.SavepointV0;
import org.apache.flink.runtime.deployment.TaskDeploymentDescriptor;
import org.apache.flink.runtime.execution.SuppressRestartsException;
import org.apache.flink.runtime.instance.ActorGateway;
@@ -84,7 +84,6 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
-import java.util.Map;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -220,9 +219,9 @@ public class SavepointITCase extends TestLogger {
new RequestSavepoint(savepointPath),
deadline.timeLeft());
- CompletedCheckpoint savepoint = ((ResponseSavepoint) Await.result(
+ SavepointV0 savepoint = (SavepointV0) ((ResponseSavepoint) Await.result(
savepointFuture, deadline.timeLeft())).savepoint();
- LOG.info("Retrieved savepoint: " + savepoint + ".");
+ LOG.info("Retrieved savepoint: " + savepointPath + ".");
// Shut down the Flink cluster (thereby canceling the job)
LOG.info("Shutting down Flink cluster.");
@@ -321,12 +320,11 @@ public class SavepointITCase extends TestLogger {
// Verify that all tasks, which are part of the savepoint
// have a matching task deployment descriptor.
- for (Map.Entry<JobVertexID, TaskState> entry: savepoint.getTaskStates().entrySet()) {
- Collection<TaskDeploymentDescriptor> taskTdds = tdds.get(entry.getKey());
- TaskState taskState = entry.getValue();
+ for (TaskState taskState : savepoint.getTaskStates()) {
+ Collection<TaskDeploymentDescriptor> taskTdds = tdds.get(taskState.getJobVertexID());
errMsg = "Missing task for savepoint state for operator "
- + entry.getKey() + ".";
+ + taskState.getJobVertexID() + ".";
assertTrue(errMsg, taskTdds.size() > 0);
assertEquals(taskState.getNumberCollectedStates(), taskTdds.size());
@@ -360,7 +358,7 @@ public class SavepointITCase extends TestLogger {
// The checkpoint files
List<File> checkpointFiles = new ArrayList<>();
- for (TaskState stateForTaskGroup : savepoint.getTaskStates().values()) {
+ for (TaskState stateForTaskGroup : savepoint.getTaskStates()) {
for (SubtaskState subtaskState : stateForTaskGroup.getStates()) {
StreamTaskStateList taskStateList = (StreamTaskStateList) subtaskState.getState()
.deserializeValue(ClassLoader.getSystemClassLoader());
@@ -389,7 +387,7 @@ public class SavepointITCase extends TestLogger {
// All savepoints should have been cleaned up
errMsg = "Savepoints directory not cleaned up properly: " +
Arrays.toString(savepointDir.listFiles()) + ".";
- assertNull(errMsg, savepointDir.listFiles());
+ assertEquals(errMsg, 0, savepointDir.listFiles().length);
// - Verification END ---------------------------------------------
}
@@ -505,9 +503,8 @@ public class SavepointITCase extends TestLogger {
new RequestSavepoint(savepointPath),
deadline.timeLeft());
- CompletedCheckpoint savepoint = ((ResponseSavepoint) Await.result(
- savepointFuture, deadline.timeLeft())).savepoint();
- LOG.info("Retrieved savepoint: " + savepoint + ".");
+ Await.ready(savepointFuture, deadline.timeLeft());
+ LOG.info("Retrieved savepoint: " + savepointPath + ".");
// Shut down the Flink cluster (thereby canceling the job)
LOG.info("Shutting down Flink cluster.");
@@ -644,9 +641,9 @@ public class SavepointITCase extends TestLogger {
new RequestSavepoint(savepointPath),
deadline.timeLeft());
- CompletedCheckpoint savepoint = ((ResponseSavepoint) Await.result(
+ SavepointV0 savepoint = (SavepointV0) ((ResponseSavepoint) Await.result(
savepointFuture, deadline.timeLeft())).savepoint();
- LOG.info("Retrieved savepoint: " + savepoint + ".");
+ LOG.info("Retrieved savepoint: " + savepointPath + ".");
// Cancel the job
LOG.info("Cancelling job " + jobId + ".");
@@ -661,7 +658,7 @@ public class SavepointITCase extends TestLogger {
assertTrue((Boolean) Await.result(removedRespFuture, deadline.timeLeft()));
// Check that all checkpoint files have been removed
- for (TaskState stateForTaskGroup : savepoint.getTaskStates().values()) {
+ for (TaskState stateForTaskGroup : savepoint.getTaskStates()) {
for (SubtaskState subtaskState : stateForTaskGroup.getStates()) {
StreamTaskStateList taskStateList = (StreamTaskStateList) subtaskState.getState()
.deserializeValue(ClassLoader.getSystemClassLoader());
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-tests/src/test/java/org/apache/flink/test/classloading/ClassLoaderITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/classloading/ClassLoaderITCase.java b/flink-tests/src/test/java/org/apache/flink/test/classloading/ClassLoaderITCase.java
index eff8a0e..cdc7a80 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/classloading/ClassLoaderITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/classloading/ClassLoaderITCase.java
@@ -27,7 +27,7 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.Path;
import org.apache.flink.runtime.blob.BlobClient;
import org.apache.flink.runtime.blob.BlobKey;
-import org.apache.flink.runtime.checkpoint.SavepointStoreFactory;
+import org.apache.flink.runtime.checkpoint.savepoint.SavepointStoreFactory;
import org.apache.flink.runtime.client.JobStatusMessage;
import org.apache.flink.runtime.instance.ActorGateway;
import org.apache.flink.runtime.messages.JobManagerMessages;
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-yarn-tests/src/test/scala/org/apache/flink/yarn/TestingYarnJobManager.scala
----------------------------------------------------------------------
diff --git a/flink-yarn-tests/src/test/scala/org/apache/flink/yarn/TestingYarnJobManager.scala b/flink-yarn-tests/src/test/scala/org/apache/flink/yarn/TestingYarnJobManager.scala
index 6a6faf5..5512ad7 100644
--- a/flink-yarn-tests/src/test/scala/org/apache/flink/yarn/TestingYarnJobManager.scala
+++ b/flink-yarn-tests/src/test/scala/org/apache/flink/yarn/TestingYarnJobManager.scala
@@ -22,7 +22,8 @@ import java.util.concurrent.ExecutorService
import akka.actor.ActorRef
import org.apache.flink.configuration.Configuration
-import org.apache.flink.runtime.checkpoint.{SavepointStore, CheckpointRecoveryFactory}
+import org.apache.flink.runtime.checkpoint.CheckpointRecoveryFactory
+import org.apache.flink.runtime.checkpoint.savepoint.SavepointStore
import org.apache.flink.metrics.MetricRegistry
import org.apache.flink.runtime.execution.librarycache.BlobLibraryCacheManager
import org.apache.flink.runtime.executiongraph.restart.RestartStrategyFactory
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-yarn/src/main/scala/org/apache/flink/yarn/YarnJobManager.scala
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/scala/org/apache/flink/yarn/YarnJobManager.scala b/flink-yarn/src/main/scala/org/apache/flink/yarn/YarnJobManager.scala
index a2ac029..c772a25 100644
--- a/flink-yarn/src/main/scala/org/apache/flink/yarn/YarnJobManager.scala
+++ b/flink-yarn/src/main/scala/org/apache/flink/yarn/YarnJobManager.scala
@@ -25,7 +25,8 @@ import akka.actor.ActorRef
import org.apache.flink.api.common.JobID
import org.apache.flink.configuration.{Configuration => FlinkConfiguration, ConfigConstants}
import org.apache.flink.metrics.MetricRegistry
-import org.apache.flink.runtime.checkpoint.{SavepointStore, CheckpointRecoveryFactory}
+import org.apache.flink.runtime.checkpoint.CheckpointRecoveryFactory
+import org.apache.flink.runtime.checkpoint.savepoint.SavepointStore
import org.apache.flink.runtime.clusterframework.ApplicationStatus
import org.apache.flink.runtime.executiongraph.restart.RestartStrategyFactory
import org.apache.flink.runtime.clusterframework.messages._
@@ -38,7 +39,6 @@ import org.apache.flink.runtime.execution.librarycache.BlobLibraryCacheManager
import org.apache.flink.runtime.instance.InstanceManager
import org.apache.flink.runtime.jobmanager.scheduler.{Scheduler => FlinkScheduler}
-
import scala.concurrent.duration._
import scala.language.postfixOps
[3/3] flink git commit: [FLINK-4067] [runtime] Add savepoint headers
Posted by uc...@apache.org.
[FLINK-4067] [runtime] Add savepoint headers
Savepoints were previously persisted without any meta data using default
Java serialization of `CompletedCheckpoint`. This commit introduces a
savepoint interface with version-specific serializers and stores
savepoints with meta data.
Savepoints expose a version number and a Collection<TaskState> for
savepoint restore.
Currently, there is only one savepoint version:
SavepointV0 (Flink 1.1): This is the current savepoint version, which
holds a reference to the Checkpoint task state collection, but is
serialized with a custom serializater not relying on default Java
serialization. Therefore, it should not happen again that we need to
stick to certain classes in future Flink versions.
The savepoints are stored in `FsSavepointStore` with the following
format:
MagicNumber SavepointVersion Savepoint
- MagicNumber => int
- SavepointVersion => int (returned by Savepoint#getVersion())
- Savepoint => bytes (serialized via version-specific SavepointSerializer)
The header is minimal (magic number, version). All savepoint-specific
meta data can be moved to the savepoint itself. This is also were we
would have to add new meta data in future versions, allowing us to
differentiate between different savepoint versions when we change the
serialization stack.
All savepoint related classes have been moved from checkpoint to a new
sub package `checkpoint.savepoint`.
This closes #2194.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/f0ac261a
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/f0ac261a
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/f0ac261a
Branch: refs/heads/master
Commit: f0ac261add2eb184358d5b89bc90b13aef7b267d
Parents: 110bba3
Author: Ufuk Celebi <uc...@apache.org>
Authored: Thu Jun 30 16:16:41 2016 +0200
Committer: Ufuk Celebi <uc...@apache.org>
Committed: Tue Jul 26 11:30:43 2016 +0200
----------------------------------------------------------------------
.../org/apache/flink/util/SerializedValue.java | 24 +-
.../checkpoint/CheckpointCoordinator.java | 2 +-
.../checkpoint/CheckpointRecoveryFactory.java | 2 +-
.../runtime/checkpoint/CompletedCheckpoint.java | 12 +-
.../checkpoint/FileSystemStateStore.java | 116 --
.../runtime/checkpoint/HeapStateStore.java | 84 --
.../runtime/checkpoint/PendingCheckpoint.java | 4 +-
.../checkpoint/SavepointCoordinator.java | 383 ------
.../SavepointCoordinatorDeActivator.java | 63 -
.../runtime/checkpoint/SavepointStore.java | 67 -
.../checkpoint/SavepointStoreFactory.java | 98 --
.../StandaloneCheckpointRecoveryFactory.java | 2 +-
.../flink/runtime/checkpoint/StateStore.java | 60 -
.../flink/runtime/checkpoint/SubtaskState.java | 2 +-
.../ZooKeeperCheckpointRecoveryFactory.java | 2 +-
.../checkpoint/savepoint/FsSavepointStore.java | 190 +++
.../savepoint/HeapSavepointStore.java | 158 +++
.../runtime/checkpoint/savepoint/Savepoint.java | 80 ++
.../savepoint/SavepointCoordinator.java | 387 ++++++
.../SavepointCoordinatorDeActivator.java | 64 +
.../savepoint/SavepointSerializer.java | 54 +
.../savepoint/SavepointSerializers.java | 73 ++
.../checkpoint/savepoint/SavepointStore.java | 71 ++
.../savepoint/SavepointStoreFactory.java | 97 ++
.../checkpoint/savepoint/SavepointV0.java | 96 ++
.../savepoint/SavepointV1Serializer.java | 186 +++
.../runtime/executiongraph/ExecutionGraph.java | 11 +-
.../flink/runtime/jobgraph/JobVertexID.java | 6 +-
.../flink/runtime/util/ZooKeeperUtils.java | 2 +-
.../flink/runtime/jobmanager/JobManager.scala | 26 +-
.../testingUtils/TestingJobManager.scala | 3 +-
.../testingUtils/TestingJobManagerLike.scala | 2 +-
.../TestingJobManagerMessages.scala | 3 +-
.../checkpoint/AbstractStateStoreTest.java | 63 -
.../CompletedCheckpointStoreTest.java | 26 +-
...ExecutionGraphCheckpointCoordinatorTest.java | 4 +-
.../checkpoint/FileSystemStateStoreTest.java | 68 --
.../runtime/checkpoint/HeapStateStoreTest.java | 44 -
.../checkpoint/SavepointCoordinatorTest.java | 1140 ------------------
.../checkpoint/SavepointStoreFactoryTest.java | 96 --
.../StandaloneCompletedCheckpointStoreTest.java | 4 +-
...ZooKeeperCompletedCheckpointStoreITCase.java | 10 +-
.../savepoint/FsSavepointStoreTest.java | 235 ++++
.../savepoint/HeapSavepointStoreTest.java | 25 +
.../savepoint/SavepointCoordinatorTest.java | 1119 +++++++++++++++++
.../savepoint/SavepointStoreFactoryTest.java | 91 ++
.../savepoint/SavepointV01SerializerTest.java | 52 +
.../checkpoint/savepoint/SavepointV01Test.java | 81 ++
.../jobmanager/JobManagerHARecoveryTest.java | 7 +-
.../JobManagerLeaderElectionTest.java | 4 +-
.../runtime/jobmanager/JobManagerITCase.scala | 3 +-
.../test/checkpointing/SavepointITCase.java | 31 +-
.../test/classloading/ClassLoaderITCase.java | 2 +-
.../flink/yarn/TestingYarnJobManager.scala | 3 +-
.../org/apache/flink/yarn/YarnJobManager.scala | 4 +-
55 files changed, 3155 insertions(+), 2387 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-core/src/main/java/org/apache/flink/util/SerializedValue.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/util/SerializedValue.java b/flink-core/src/main/java/org/apache/flink/util/SerializedValue.java
index 4612c91..d91c14e 100644
--- a/flink-core/src/main/java/org/apache/flink/util/SerializedValue.java
+++ b/flink-core/src/main/java/org/apache/flink/util/SerializedValue.java
@@ -43,31 +43,35 @@ public class SerializedValue<T> implements java.io.Serializable {
/** The serialized data */
private final byte[] serializedData;
+ private SerializedValue(byte[] serializedData) {
+ Preconditions.checkNotNull(serializedData, "Serialized data");
+ this.serializedData = serializedData;
+ }
public SerializedValue(T value) throws IOException {
this.serializedData = value == null ? null : InstantiationUtil.serializeObject(value);
}
-
@SuppressWarnings("unchecked")
public T deserializeValue(ClassLoader loader) throws IOException, ClassNotFoundException {
- if (loader == null) {
- throw new NullPointerException("No classloader has been passed");
- }
-
+ Preconditions.checkNotNull(loader, "No classloader has been passed");
return serializedData == null ? null : (T) InstantiationUtil.deserializeObject(serializedData, loader);
}
/**
- * Gets the size of the serialized state.
- * @return The size of the serialized state.
+ * Returns the serialized value or <code>null</code> if no value is set.
+ *
+ * @return Serialized data.
*/
- public int getSizeOfSerializedState() {
- return serializedData.length;
+ public byte[] getByteArray() {
+ return serializedData;
}
- // --------------------------------------------------------------------------------------------
+ public static <T> SerializedValue<T> fromBytes(byte[] serializedData) {
+ return new SerializedValue<T>(serializedData);
+ }
+ // --------------------------------------------------------------------------------------------
@Override
public int hashCode() {
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/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 b259b91..6aaa014 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
@@ -684,7 +684,7 @@ public class CheckpointCoordinator {
message.getStateSize(),
null)) { // TODO: Give KV-state to the acknowledgeTask method
if (checkpoint.isFullyAcknowledged()) {
- completed = checkpoint.toCompletedCheckpoint();
+ completed = checkpoint.finalizeCheckpoint();
completedCheckpointStore.addCheckpoint(completed);
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointRecoveryFactory.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointRecoveryFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointRecoveryFactory.java
index aa6e94b..0c7dfa7 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointRecoveryFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointRecoveryFactory.java
@@ -47,7 +47,7 @@ public interface CheckpointRecoveryFactory {
* @param userClassLoader User code class loader of the job
* @return {@link CompletedCheckpointStore} instance for the job
*/
- CompletedCheckpointStore createCompletedCheckpoints(JobID jobId, ClassLoader userClassLoader)
+ CompletedCheckpointStore createCheckpointStore(JobID jobId, ClassLoader userClassLoader)
throws Exception;
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CompletedCheckpoint.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CompletedCheckpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CompletedCheckpoint.java
index 3e1db02..ecc67b4 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CompletedCheckpoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CompletedCheckpoint.java
@@ -54,7 +54,7 @@ public class CompletedCheckpoint implements Serializable {
long completionTimestamp,
Map<JobVertexID, TaskState> taskStates) {
- this.job = job;
+ this.job = Preconditions.checkNotNull(job);
this.checkpointID = checkpointID;
this.timestamp = timestamp;
this.duration = completionTimestamp - timestamp;
@@ -98,13 +98,11 @@ public class CompletedCheckpoint implements Serializable {
// --------------------------------------------------------------------------------------------
public void discard(ClassLoader userClassLoader) throws Exception {
- try {
- for (TaskState state : taskStates.values()) {
- state.discard(userClassLoader);
- }
- } finally {
- taskStates.clear();
+ for (TaskState state: taskStates.values()) {
+ state.discard(userClassLoader);
}
+
+ taskStates.clear();
}
// --------------------------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/FileSystemStateStore.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/FileSystemStateStore.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/FileSystemStateStore.java
deleted file mode 100644
index 4ab0b54..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/FileSystemStateStore.java
+++ /dev/null
@@ -1,116 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.checkpoint;
-
-import org.apache.flink.core.fs.FSDataOutputStream;
-import org.apache.flink.core.fs.FileSystem;
-import org.apache.flink.core.fs.Path;
-import org.apache.flink.runtime.state.filesystem.FileSerializableStateHandle;
-import org.apache.flink.util.FileUtils;
-
-import java.io.IOException;
-import java.io.ObjectOutputStream;
-import java.io.Serializable;
-
-import static org.apache.flink.util.Preconditions.checkNotNull;
-
-/**
- * {@link FileSystem} backed {@link StateStore}.
- *
- * @param <T> Type of state
- */
-class FileSystemStateStore<T extends Serializable> implements StateStore<T> {
-
- private final Path rootPath;
-
- private final String prefix;
-
- private final FileSystem fileSystem;
-
- FileSystemStateStore(String rootPath, String prefix) throws IOException {
- this(new Path(rootPath), prefix);
- }
-
- FileSystemStateStore(Path rootPath, String prefix) throws IOException {
- this.rootPath = checkNotNull(rootPath, "Root path");
- this.prefix = checkNotNull(prefix, "Prefix");
-
- this.fileSystem = FileSystem.get(rootPath.toUri());
- }
-
- @Override
- public String putState(T state) throws Exception {
- Exception latestException = null;
-
- for (int attempt = 0; attempt < 10; attempt++) {
- Path filePath = new Path(rootPath, FileUtils.getRandomFilename(prefix));
- FSDataOutputStream outStream;
- try {
- outStream = fileSystem.create(filePath, false);
- }
- catch (Exception e) {
- latestException = e;
- continue;
- }
-
- try (ObjectOutputStream os = new ObjectOutputStream(outStream)) {
- os.writeObject(state);
- }
-
- return filePath.toString();
- }
-
- throw new IOException("Failed to open file output stream", latestException);
- }
-
- @Override
- @SuppressWarnings("unchecked")
- public T getState(String path) throws Exception {
- Path filePath = new Path(path);
-
- if (FileSystem.get(filePath.toUri()).exists(filePath)) {
- return (T) new FileSerializableStateHandle<>(filePath)
- .getState(ClassLoader.getSystemClassLoader());
- }
- else {
- throw new IllegalArgumentException("Invalid path '" + filePath.toUri() + "'.");
- }
- }
-
- @Override
- public void disposeState(String path) throws Exception {
- Path filePath = new Path(path);
-
- if (!fileSystem.delete(filePath, true)) {
- throw new IllegalArgumentException("Invalid path '" + filePath.toUri() + "'.");
- }
-
- // send a call to delete the directory containing the file. this will
- // fail (and be ignored) when some files still exist
- try {
- fileSystem.delete(filePath.getParent(), false);
- }
- catch (IOException ignored) {
- }
- }
-
- Path getRootPath() {
- return rootPath;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/HeapStateStore.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/HeapStateStore.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/HeapStateStore.java
deleted file mode 100644
index c679d50..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/HeapStateStore.java
+++ /dev/null
@@ -1,84 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.checkpoint;
-
-import java.io.Serializable;
-import java.util.Collection;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.atomic.AtomicInteger;
-
-import static org.apache.flink.util.Preconditions.checkNotNull;
-
-/**
- * Java heap backed {@link StateStore}.
- *
- * @param <T> Type of state
- */
-public class HeapStateStore<T extends Serializable> implements StateStore<T> {
-
- private final ConcurrentMap<String, T> stateMap = new ConcurrentHashMap<>();
-
- private final AtomicInteger idCounter = new AtomicInteger();
-
- @Override
- public String putState(T state) throws Exception {
- checkNotNull(state, "State");
-
- String key = "jobmanager://savepoints/" + idCounter.incrementAndGet();
- stateMap.put(key, state);
- return key;
- }
-
- @Override
- public T getState(String path) throws Exception {
- T state = stateMap.get(path);
-
- if (state != null) {
- return state;
- }
- else {
- throw new IllegalArgumentException("Invalid path '" + path + "'.");
- }
- }
-
- @Override
- public void disposeState(String path) throws Exception {
- T state = stateMap.remove(path);
-
- if (state == null) {
- throw new IllegalArgumentException("Invalid path '" + path + "'.");
- }
- }
-
- /**
- * Returns all stored state.
- */
- Collection<T> getAll() {
- return stateMap.values();
- }
-
- /**
- * Clears all stored state.
- */
- void clearAll() {
- stateMap.clear();
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/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 8c9069c..cd1f6c4 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
@@ -53,7 +53,6 @@ public class PendingCheckpoint {
private int numAcknowledgedTasks;
private boolean discarded;
- private ClassLoader userClassLoader;
// --------------------------------------------------------------------------------------------
@@ -110,7 +109,7 @@ public class PendingCheckpoint {
return discarded;
}
- public CompletedCheckpoint toCompletedCheckpoint() throws Exception {
+ public CompletedCheckpoint finalizeCheckpoint() throws Exception {
synchronized (lock) {
if (discarded) {
throw new IllegalStateException("pending checkpoint is discarded");
@@ -196,7 +195,6 @@ public class PendingCheckpoint {
* Discards the pending checkpoint, releasing all held resources.
*/
public void discard(ClassLoader userClassLoader) throws Exception {
- this.userClassLoader = userClassLoader;
dispose(userClassLoader, true);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/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
deleted file mode 100644
index e58e277..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SavepointCoordinator.java
+++ /dev/null
@@ -1,383 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.checkpoint;
-
-import akka.actor.ActorSystem;
-import akka.actor.Props;
-import org.apache.flink.api.common.JobID;
-import org.apache.flink.runtime.checkpoint.stats.CheckpointStatsTracker;
-import org.apache.flink.runtime.execution.ExecutionState;
-import org.apache.flink.runtime.executiongraph.Execution;
-import org.apache.flink.runtime.executiongraph.ExecutionJobVertex;
-import org.apache.flink.runtime.executiongraph.ExecutionVertex;
-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.runtime.state.StateHandle;
-import org.apache.flink.util.SerializedValue;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import scala.concurrent.Future;
-import scala.concurrent.Promise;
-
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.UUID;
-import java.util.concurrent.ConcurrentHashMap;
-
-import static org.apache.flink.util.Preconditions.checkNotNull;
-
-/**
- * The savepoint coordinator is a slightly modified variant of the regular
- * checkpoint coordinator. Checkpoints are not triggered periodically, but
- * manually. The actual checkpointing mechanism is the same as for periodic
- * checkpoints, only the control flow is modified.
- *
- * <p>The savepoint coordinator is meant to be used as a separate coordinator
- * instance. Otherwise, there can be unwanted queueing effects like discarding
- * savepoints, because of in-progress periodic checkpoints.
- *
- * <p>The savepoint coordinator registers callbacks on the regular checkpoint
- * life-cycle and manages a map of promises, which are completed/failed as soon
- * as the trigged checkpoint is done.
- *
- * <p><strong>Important</strong>: it's necessary that both the periodic
- * checkpoint coordinator and the savepoint coordinator <em>share</em> the same
- * instance of the {@link CheckpointIDCounter} to ensure that all task managers
- * see ascending checkpoints IDs.
- */
-public class SavepointCoordinator extends CheckpointCoordinator {
-
- private static final Logger LOG = LoggerFactory.getLogger(SavepointCoordinator.class);
-
- /** Store for savepoints. */
- private StateStore<CompletedCheckpoint> savepointStore;
-
- /** Mapping from checkpoint ID to promises for savepoints. */
- private final Map<Long, Promise<String>> savepointPromises;
-
- // TODO(uce) Temporary work around to restore initial state on
- // failure during recovery. Will be superseded by FLINK-3397.
- private volatile String savepointRestorePath;
-
- public SavepointCoordinator(
- JobID jobId,
- long baseInterval,
- long checkpointTimeout,
- int numberKeyGroups,
- ExecutionVertex[] tasksToTrigger,
- ExecutionVertex[] tasksToWaitFor,
- ExecutionVertex[] tasksToCommitTo,
- ClassLoader userClassLoader,
- CheckpointIDCounter checkpointIDCounter,
- StateStore<CompletedCheckpoint> savepointStore,
- CheckpointStatsTracker statsTracker) throws Exception {
-
- super(jobId,
- baseInterval,
- checkpointTimeout,
- 0L,
- Integer.MAX_VALUE,
- numberKeyGroups,
- tasksToTrigger,
- tasksToWaitFor,
- tasksToCommitTo,
- userClassLoader,
- checkpointIDCounter,
- IgnoreCompletedCheckpointsStore.INSTANCE,
- RecoveryMode.STANDALONE,
- statsTracker);
-
- this.savepointStore = checkNotNull(savepointStore);
- this.savepointPromises = new ConcurrentHashMap<>();
- }
-
- public String getSavepointRestorePath() {
- return savepointRestorePath;
- }
-
- // ------------------------------------------------------------------------
- // Savepoint trigger and reset
- // ------------------------------------------------------------------------
-
- /**
- * Triggers a new savepoint using the current system time as the checkpoint timestamp.
- */
- public Future<String> triggerSavepoint(long timestamp) throws Exception {
- final Promise<String> promise = new scala.concurrent.impl.Promise.DefaultPromise<>();
-
- try {
- // Get the checkpoint ID up front. If we fail to trigger the checkpoint,
- // the ID will have changed, but this is OK as long as the checkpoint ID
- // generates ascending IDs.
- final long checkpointId = getAndIncrementCheckpointId();
-
- if (checkpointId == -1) {
- throw new IllegalStateException("Failed to get checkpoint Id");
- }
-
- LOG.info("Triggering savepoint with ID " + checkpointId);
-
- // Important: make sure to add the promise to the map before calling
- // any methods that might trigger callbacks, which require the promise.
- // Otherwise, the might be race conditions.
- if (savepointPromises.put(checkpointId, promise) == null) {
- boolean success = false;
-
- try {
- // All good. The future will be completed as soon as the
- // triggered checkpoint is done.
- success = triggerCheckpoint(timestamp, checkpointId);
- }
- finally {
- if (!success) {
- savepointPromises.remove(checkpointId);
- promise.failure(new Exception("Failed to trigger savepoint"));
- }
- }
- }
- else {
- throw new IllegalStateException("Duplicate checkpoint ID");
- }
- }
- catch (Throwable t) {
- promise.failure(new Exception("Failed to trigger savepoint", t));
- }
-
- return promise.future();
- }
-
- /**
- * Resets the state of {@link Execution} instances back to the state of a savepoint.
- *
- * <p>The execution vertices need to be in state {@link ExecutionState#CREATED} when calling
- * this method. The operation might block. Make sure that calls don't block the job manager
- * actor.
- *
- * @param tasks Tasks that will possibly be reset
- * @param savepointPath The path of the savepoint to rollback to
- * @throws IllegalStateException If coordinator is shut down
- * @throws IllegalStateException If mismatch between program and savepoint state
- * @throws Exception If savepoint store failure
- */
- public void restoreSavepoint(
- Map<JobVertexID, ExecutionJobVertex> tasks,
- String savepointPath) throws Exception {
-
- checkNotNull(savepointPath, "Savepoint path");
-
- synchronized (lock) {
- if (isShutdown()) {
- throw new IllegalStateException("CheckpointCoordinator is shut down");
- }
-
- LOG.info("Rolling back to savepoint '{}'.", savepointPath);
-
- CompletedCheckpoint checkpoint = savepointStore.getState(savepointPath);
-
- LOG.info("Savepoint: {}@{}", checkpoint.getCheckpointID(), checkpoint.getTimestamp());
-
- // Set the initial state of all tasks
- LOG.debug("Rolling back individual operators.");
-
- for (Map.Entry<JobVertexID, TaskState> taskStateEntry: checkpoint.getTaskStates().entrySet()) {
- TaskState taskState = taskStateEntry.getValue();
- ExecutionJobVertex executionJobVertex = tasks.get(taskStateEntry.getKey());
-
- if (executionJobVertex != null) {
- if (executionJobVertex.getParallelism() != taskState.getParallelism()) {
- String msg = String.format("Failed to rollback to savepoint %s. " +
- "Parallelism mismatch between savepoint state and new program. " +
- "Cannot map operator %s with parallelism %d to new program with " +
- "parallelism %d. This indicates that the program has been changed " +
- "in a non-compatible way after the savepoint.",
- checkpoint,
- taskStateEntry.getKey(),
- taskState.getParallelism(),
- executionJobVertex.getParallelism());
-
- throw new IllegalStateException(msg);
- }
-
- List<Set<Integer>> keyGroupPartitions = createKeyGroupPartitions(
- numberKeyGroups,
- executionJobVertex.getParallelism());
-
- for (int i = 0; i < executionJobVertex.getTaskVertices().length; i++) {
- SubtaskState subtaskState = taskState.getState(i);
- SerializedValue<StateHandle<?>> state = null;
-
- if (subtaskState != null) {
- state = subtaskState.getState();
- }
-
- Map<Integer, SerializedValue<StateHandle<?>>> kvStateForTaskMap = taskState
- .getUnwrappedKvStates(keyGroupPartitions.get(i));
-
- Execution currentExecutionAttempt = executionJobVertex
- .getTaskVertices()[i]
- .getCurrentExecutionAttempt();
-
- currentExecutionAttempt.setInitialState(state, kvStateForTaskMap);
- }
- } else {
- 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 in a " +
- "non-compatible way after the savepoint.", checkpoint,
- taskStateEntry.getKey());
- throw new IllegalStateException(msg);
- }
- }
-
- // Reset the checkpoint ID counter
- long nextCheckpointId = checkpoint.getCheckpointID();
- checkpointIdCounter.start();
- checkpointIdCounter.setCount(nextCheckpointId + 1);
- LOG.info("Reset the checkpoint ID to {}", nextCheckpointId);
-
- if (savepointRestorePath == null) {
- savepointRestorePath = savepointPath;
- }
- }
- }
-
- // ------------------------------------------------------------------------
- // Checkpoint coordinator callbacks
- // ------------------------------------------------------------------------
-
- @Override
- protected void onShutdown() {
- // Fail all outstanding savepoint futures
- for (Promise<String> promise : savepointPromises.values()) {
- promise.failure(new Exception("Checkpoint coordinator shutdown"));
- }
- savepointPromises.clear();
- }
-
- @Override
- protected void onCancelCheckpoint(long canceledCheckpointId) {
- LOG.info("Cancelling savepoint with checkpoint ID " + canceledCheckpointId);
- Promise<String> promise = savepointPromises.remove(canceledCheckpointId);
-
- if (promise != null) {
- promise.failure(new Exception("Savepoint expired before completing"));
- }
- }
-
- @Override
- protected void onFullyAcknowledgedCheckpoint(CompletedCheckpoint checkpoint) {
- // Sanity check
- Promise<String> promise = savepointPromises.remove(checkpoint.getCheckpointID());
-
- if (promise == null) {
- LOG.info("Pending savepoint with ID " + checkpoint.getCheckpointID() + " has been " +
- "removed before receiving acknowledgment.");
- }
-
- // Sanity check
- if (promise.isCompleted()) {
- throw new IllegalStateException("Savepoint promise completed");
- }
-
- try {
- // Save the checkpoint
- String savepointPath = savepointStore.putState(checkpoint);
- promise.success(savepointPath);
- }
- catch (Exception e) {
- LOG.warn("Failed to store savepoint.", e);
- promise.failure(e);
- }
- }
-
- // ------------------------------------------------------------------------
- // Job status listener
- // ------------------------------------------------------------------------
-
- @Override
- public ActorGateway createActivatorDeactivator(
- ActorSystem actorSystem,
- UUID leaderSessionID) {
-
- synchronized (lock) {
- if (isShutdown()) {
- throw new IllegalArgumentException("Checkpoint coordinator is shut down");
- }
-
- if (getJobStatusListener() == null) {
- Props props = Props.create(
- SavepointCoordinatorDeActivator.class,
- this,
- leaderSessionID);
-
- // wrap the ActorRef in a AkkaActorGateway to support message decoration
- setJobStatusListener(new AkkaActorGateway(
- actorSystem.actorOf(props),
- leaderSessionID));
- }
-
- return getJobStatusListener();
- }
- }
-
- // ------------------------------------------------------------------------
- // Completed checkpoints
- // ------------------------------------------------------------------------
-
- private static class IgnoreCompletedCheckpointsStore implements CompletedCheckpointStore {
-
- private static final CompletedCheckpointStore INSTANCE = new IgnoreCompletedCheckpointsStore();
-
- @Override
- public void recover() throws Exception {
- }
-
- @Override
- public void addCheckpoint(CompletedCheckpoint checkpoint) throws Exception {
- }
-
- @Override
- public CompletedCheckpoint getLatestCheckpoint() throws Exception {
- return null;
- }
-
- @Override
- public void shutdown() throws Exception {
- }
-
- @Override
- public void suspend() throws Exception {
- }
-
- @Override
- public List<CompletedCheckpoint> getAllCheckpoints() throws Exception {
- return Collections.emptyList();
- }
-
- @Override
- public int getNumberOfRetainedCheckpoints() {
- return 0;
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SavepointCoordinatorDeActivator.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SavepointCoordinatorDeActivator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SavepointCoordinatorDeActivator.java
deleted file mode 100644
index 56a3738..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SavepointCoordinatorDeActivator.java
+++ /dev/null
@@ -1,63 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.checkpoint;
-
-import org.apache.flink.runtime.akka.FlinkUntypedActor;
-import org.apache.flink.runtime.jobgraph.JobStatus;
-import org.apache.flink.runtime.messages.ExecutionGraphMessages;
-import org.apache.flink.util.Preconditions;
-
-import java.util.UUID;
-
-/**
- * This actor listens to changes in the JobStatus and deactivates the
- * savepoint scheduler and discards all pending checkpoints.
- */
-public class SavepointCoordinatorDeActivator extends FlinkUntypedActor {
-
- private final CheckpointCoordinator coordinator;
- private final UUID leaderSessionID;
-
- public SavepointCoordinatorDeActivator(
- SavepointCoordinator coordinator,
- UUID leaderSessionID) {
-
- LOG.info("Create SavepointCoordinatorDeActivator");
-
- this.coordinator = Preconditions.checkNotNull(coordinator, "The checkpointCoordinator must not be null.");
- this.leaderSessionID = leaderSessionID;
- }
-
- @Override
- public void handleMessage(Object message) throws Exception {
- if (message instanceof ExecutionGraphMessages.JobStatusChanged) {
- JobStatus status = ((ExecutionGraphMessages.JobStatusChanged) message).newJobStatus();
-
- if (status != JobStatus.RUNNING) {
- // anything other than RUNNING should stop the trigger for now
- coordinator.stopCheckpointScheduler();
- }
- }
- }
-
- @Override
- public UUID getLeaderSessionID() {
- return leaderSessionID;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SavepointStore.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SavepointStore.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SavepointStore.java
deleted file mode 100644
index 2ab4d00..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SavepointStore.java
+++ /dev/null
@@ -1,67 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.checkpoint;
-
-/**
- * Simple wrapper around the state store for savepoints.
- */
-public class SavepointStore implements StateStore<CompletedCheckpoint> {
-
- private final StateStore<CompletedCheckpoint> stateStore;
-
- public SavepointStore(StateStore<CompletedCheckpoint> stateStore) {
- this.stateStore = stateStore;
- }
-
- public void start() {
- }
-
- public void stop() throws Exception {
- if (stateStore instanceof HeapStateStore) {
- HeapStateStore<CompletedCheckpoint> heapStateStore = (HeapStateStore<CompletedCheckpoint>) stateStore;
-
- try {
- for (CompletedCheckpoint savepoint : heapStateStore.getAll()) {
- savepoint.discard(ClassLoader.getSystemClassLoader());
- }
- } finally {
- heapStateStore.clearAll();
- }
- }
- }
-
- @Override
- public String putState(CompletedCheckpoint state) throws Exception {
- return stateStore.putState(state);
- }
-
- @Override
- public CompletedCheckpoint getState(String path) throws Exception {
- return stateStore.getState(path);
- }
-
- @Override
- public void disposeState(String path) throws Exception {
- stateStore.disposeState(path);
- }
-
- StateStore<CompletedCheckpoint> getStateStore() {
- return stateStore;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SavepointStoreFactory.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SavepointStoreFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SavepointStoreFactory.java
deleted file mode 100644
index 6d25e18..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SavepointStoreFactory.java
+++ /dev/null
@@ -1,98 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.checkpoint;
-
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.IllegalConfigurationException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-
-/**
- * Factory for savepoint {@link StateStore} instances.
- */
-public class SavepointStoreFactory {
-
- public static final String SAVEPOINT_BACKEND_KEY = "savepoints.state.backend";
- public static final String SAVEPOINT_DIRECTORY_KEY = "savepoints.state.backend.fs.dir";
- public static final String DEFAULT_SAVEPOINT_BACKEND = "jobmanager";
-
- public static final Logger LOG = LoggerFactory.getLogger(SavepointStoreFactory.class);
-
- /**
- * Creates a {@link SavepointStore} from the specified Configuration.
- *
- * <p>You can configure a savepoint-specific backend for the savepoints. If
- * you don't configure anything, the regular checkpoint backend will be
- * used.
- *
- * <p>The default and fallback backend is the job manager, which loses the
- * savepoint after the job manager shuts down.
- *
- * @param config The configuration to parse the savepoint backend configuration.
- * @return A savepoint store.
- */
- public static SavepointStore createFromConfig(
- Configuration config) throws Exception {
-
- // Try a the savepoint-specific configuration first.
- String savepointBackend = config.getString(SAVEPOINT_BACKEND_KEY, DEFAULT_SAVEPOINT_BACKEND);
-
- if (savepointBackend == null) {
- LOG.info("No savepoint state backend configured. " +
- "Using job manager savepoint state backend.");
- return createJobManagerSavepointStore();
- } else if (savepointBackend.equals("jobmanager")) {
- LOG.info("Using job manager savepoint state backend.");
- return createJobManagerSavepointStore();
- } else if (savepointBackend.equals("filesystem")) {
- String rootPath = config.getString(SAVEPOINT_DIRECTORY_KEY, null);
-
- if (rootPath == null) {
- throw new IllegalConfigurationException("Using filesystem as savepoint state backend, " +
- "but did not specify directory. Please set the " +
- "following configuration key: '" + SAVEPOINT_DIRECTORY_KEY +
- "' (e.g. " + SAVEPOINT_DIRECTORY_KEY + ": hdfs:///flink/savepoints/). " +
- "Falling back to job manager savepoint backend.");
- } else {
- LOG.info("Using filesystem savepoint backend (root path: {}).", rootPath);
-
- return createFileSystemSavepointStore(rootPath);
- }
- } else {
- throw new IllegalConfigurationException("Unexpected savepoint backend " +
- "configuration '" + savepointBackend + "'. " +
- "Falling back to job manager savepoint state backend.");
- }
- }
-
- // ------------------------------------------------------------------------
- // Savepoint stores
- // ------------------------------------------------------------------------
-
- private static SavepointStore createJobManagerSavepointStore() {
- return new SavepointStore(new HeapStateStore<CompletedCheckpoint>());
- }
-
- private static SavepointStore createFileSystemSavepointStore(String rootPath) throws IOException {
- return new SavepointStore(new FileSystemStateStore<CompletedCheckpoint>(rootPath, "savepoint-"));
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StandaloneCheckpointRecoveryFactory.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StandaloneCheckpointRecoveryFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StandaloneCheckpointRecoveryFactory.java
index 324a0be..05f9e77 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StandaloneCheckpointRecoveryFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StandaloneCheckpointRecoveryFactory.java
@@ -37,7 +37,7 @@ public class StandaloneCheckpointRecoveryFactory implements CheckpointRecoveryFa
}
@Override
- public CompletedCheckpointStore createCompletedCheckpoints(JobID jobId, ClassLoader userClassLoader)
+ public CompletedCheckpointStore createCheckpointStore(JobID jobId, ClassLoader userClassLoader)
throws Exception {
return new StandaloneCompletedCheckpointStore(CheckpointRecoveryFactory
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StateStore.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StateStore.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StateStore.java
deleted file mode 100644
index 2ceceba..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StateStore.java
+++ /dev/null
@@ -1,60 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.checkpoint;
-
-import java.io.Serializable;
-
-/**
- * A simple {@link org.apache.flink.runtime.state.StateBackend} backend object
- * store interface.
- *
- * <p><strong>Important:</strong> This does not work with user-specified classes.
- *
- * @param <T> The type of objects to store.
- */
-public interface StateStore<T extends Serializable> {
-
- /**
- * Puts an object into the store and returns the (logical) path to it.
- *
- * <p>The stored state can be retrieved or discarded by specifying the returned path.
- *
- * @param state The object to store.
- * @return The path to retrieve or discard the object again.
- */
- String putState(T state) throws Exception;
-
- /**
- * Returns the object identified by the given (logical) path.
- *
- * @param path The path of the object to retrieve.
- * @return The object stored under the specified path.
- * @throws IllegalArgumentException If the path does not identify a valid object.
- */
- T getState(String path) throws Exception;
-
- /**
- * Disposes the state identified by the given (logical) path.
- *
- * @param path The path of the object to discard.
- * @throws IllegalArgumentException If the path does not identify a valid object.
- */
- void disposeState(String path) throws Exception;
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SubtaskState.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SubtaskState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SubtaskState.java
index 9fcafe8..f326eaa 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SubtaskState.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SubtaskState.java
@@ -111,6 +111,6 @@ public class SubtaskState implements Serializable {
@Override
public String toString() {
- return String.format("StateForTask(Size: %d, Duration: %d, State: %s)", stateSize, duration, state);
+ return String.format("SubtaskState(Size: %d, Duration: %d, State: %s)", stateSize, duration, state);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ZooKeeperCheckpointRecoveryFactory.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ZooKeeperCheckpointRecoveryFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ZooKeeperCheckpointRecoveryFactory.java
index f9d6d3e..dcd6260 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ZooKeeperCheckpointRecoveryFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ZooKeeperCheckpointRecoveryFactory.java
@@ -51,7 +51,7 @@ public class ZooKeeperCheckpointRecoveryFactory implements CheckpointRecoveryFac
}
@Override
- public CompletedCheckpointStore createCompletedCheckpoints(JobID jobId, ClassLoader userClassLoader)
+ public CompletedCheckpointStore createCheckpointStore(JobID jobId, ClassLoader userClassLoader)
throws Exception {
return ZooKeeperUtils.createCompletedCheckpoints(client, config, jobId,
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/FsSavepointStore.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/FsSavepointStore.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/FsSavepointStore.java
new file mode 100644
index 0000000..6efc01e
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/FsSavepointStore.java
@@ -0,0 +1,190 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.checkpoint.savepoint;
+
+import org.apache.flink.core.fs.FSDataInputStream;
+import org.apache.flink.core.fs.FSDataOutputStream;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.core.memory.DataInputViewStreamWrapper;
+import org.apache.flink.util.FileUtils;
+import org.apache.flink.util.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * A file system based {@link SavepointStore}.
+ *
+ * <p>Stored savepoints have the following format:
+ * <pre>
+ * MagicNumber SavepointVersion Savepoint
+ * - MagicNumber => int
+ * - SavepointVersion => int (returned by Savepoint#getVersion())
+ * - Savepoint => bytes (serialized via version-specific SavepointSerializer)
+ * </pre>
+ */
+public class FsSavepointStore implements SavepointStore {
+
+ private static final Logger LOG = LoggerFactory.getLogger(FsSavepointStore.class);
+
+ /** Magic number for sanity checks against stored savepoints. */
+ int MAGIC_NUMBER = 0x4960672d;
+
+ /** Root path for savepoints. */
+ private final Path rootPath;
+
+ /** Prefix for savepoint files. */
+ private final String prefix;
+
+ /** File system to use for file access. */
+ private final FileSystem fileSystem;
+
+ /**
+ * Creates a new file system based {@link SavepointStore}.
+ *
+ * @param rootPath Root path for savepoints
+ * @param prefix Prefix for savepoint files
+ * @throws IOException On failure to access root path
+ */
+ FsSavepointStore(String rootPath, String prefix) throws IOException {
+ this.rootPath = new Path(checkNotNull(rootPath, "Root path"));
+ this.prefix = checkNotNull(prefix, "Prefix");
+
+ this.fileSystem = FileSystem.get(this.rootPath.toUri());
+ }
+
+ @Override
+ public <T extends Savepoint> String storeSavepoint(T savepoint) throws IOException {
+ Preconditions.checkNotNull(savepoint, "Savepoint");
+
+ Exception latestException = null;
+ Path path = null;
+ FSDataOutputStream fdos = null;
+
+ // Try to create a FS output stream
+ for (int attempt = 0; attempt < 10; attempt++) {
+ path = new Path(rootPath, FileUtils.getRandomFilename(prefix));
+ try {
+ fdos = fileSystem.create(path, false);
+ break;
+ } catch (Exception e) {
+ latestException = e;
+ }
+ }
+
+ if (fdos == null) {
+ throw new IOException("Failed to create file output stream at " + path, latestException);
+ }
+
+ boolean success = false;
+ try (DataOutputStream dos = new DataOutputStream(fdos)) {
+ // Write header
+ dos.writeInt(MAGIC_NUMBER);
+ dos.writeInt(savepoint.getVersion());
+
+ // Write savepoint
+ SavepointSerializer<T> serializer = SavepointSerializers.getSerializer(savepoint);
+ serializer.serialize(savepoint, dos);
+ success = true;
+ } finally {
+ if (!success && fileSystem.exists(path)) {
+ if (!fileSystem.delete(path, true)) {
+ LOG.warn("Failed to delete file " + path + " after failed write.");
+ }
+ }
+ }
+
+ return path.toString();
+ }
+
+ @Override
+ public Savepoint loadSavepoint(String path) throws IOException {
+ Preconditions.checkNotNull(path, "Path");
+
+ try (DataInputStream dis = new DataInputViewStreamWrapper(createFsInputStream(new Path(path)))) {
+ int magicNumber = dis.readInt();
+
+ if (magicNumber == MAGIC_NUMBER) {
+ int version = dis.readInt();
+
+ SavepointSerializer<?> serializer = SavepointSerializers.getSerializer(version);
+ return serializer.deserialize(dis);
+ } else {
+ throw new RuntimeException("Unexpected magic number. This is most likely " +
+ "caused by trying to load a Flink 1.0 savepoint. You cannot load a " +
+ "savepoint triggered by Flink 1.0 with this version of Flink. If it is " +
+ "_not_ a Flink 1.0 savepoint, this error indicates that the specified " +
+ "file is not a proper savepoint or the file has been corrupted.");
+ }
+ }
+ }
+
+ @Override
+ public void disposeSavepoint(String path, ClassLoader classLoader) throws Exception {
+ Preconditions.checkNotNull(path, "Path");
+ Preconditions.checkNotNull(classLoader, "Class loader");
+
+ try {
+ Savepoint savepoint = loadSavepoint(path);
+ savepoint.dispose(classLoader);
+
+ Path filePath = new Path(path);
+
+ if (fileSystem.exists(filePath)) {
+ if (!fileSystem.delete(filePath, true)) {
+ throw new IOException("Failed to delete " + filePath + ".");
+ }
+ } else {
+ throw new IllegalArgumentException("Invalid path '" + filePath.toUri() + "'.");
+ }
+ } catch (Throwable t) {
+ throw new IOException("Failed to dispose savepoint " + path + ".", t);
+ }
+ }
+
+ @Override
+ public void shutdown() throws Exception {
+ // Nothing to do, because the savepoint life-cycle is independent of
+ // the cluster life-cycle.
+ }
+
+ private FSDataInputStream createFsInputStream(Path path) throws IOException {
+ if (fileSystem.exists(path)) {
+ return fileSystem.open(path);
+ } else {
+ throw new IllegalArgumentException("Invalid path '" + path.toUri() + "'.");
+ }
+ }
+
+ /**
+ * Returns the savepoint root path.
+ *
+ * @return Savepoint root path
+ */
+ Path getRootPath() {
+ return rootPath;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/HeapSavepointStore.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/HeapSavepointStore.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/HeapSavepointStore.java
new file mode 100644
index 0000000..cf30f5f
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/HeapSavepointStore.java
@@ -0,0 +1,158 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.checkpoint.savepoint;
+
+import org.apache.flink.util.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Heap-backed savepoint store.
+ *
+ * <p>The life-cycle of savepoints is bound to the life-cycle of the cluster.
+ */
+public class HeapSavepointStore implements SavepointStore {
+
+ private static final Logger LOG = LoggerFactory.getLogger(HeapSavepointStore.class);
+
+ private final Object shutDownLock = new Object();
+
+ /** Stored savepoints. */
+ private final Map<String, Savepoint> savepoints = new HashMap<>(1);
+
+ /** ID counter to identify savepoints. */
+ private final AtomicInteger currentId = new AtomicInteger();
+
+ /** Flag indicating whether state store has been shut down. */
+ private boolean shutDown;
+
+ /** Shut down hook. */
+ private final Thread shutdownHook;
+
+ /**
+ * Creates a heap-backed savepoint store.
+ *
+ * <p>Savepoint are discarded on {@link #shutdown()}.
+ */
+ public HeapSavepointStore() {
+ this.shutdownHook = new Thread(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ shutdown();
+ } catch (Throwable t) {
+ LOG.warn("Failure during shut down hook.", t);
+ }
+ }
+ });
+
+ try {
+ Runtime.getRuntime().addShutdownHook(shutdownHook);
+ } catch (IllegalStateException ignored) {
+ // JVM is already shutting down
+ } catch (Throwable t) {
+ LOG.warn("Failed to register shutdown hook.");
+ }
+ }
+
+ @Override
+ public <T extends Savepoint> String storeSavepoint(T savepoint) throws IOException {
+ Preconditions.checkNotNull(savepoint, "Savepoint");
+
+ synchronized (shutDownLock) {
+ if (shutDown) {
+ throw new IllegalStateException("Shut down");
+ } else {
+ String path = "jobmanager://savepoints/" + currentId.incrementAndGet();
+ savepoints.put(path, savepoint);
+ return path;
+ }
+ }
+ }
+
+ @Override
+ public Savepoint loadSavepoint(String path) throws IOException {
+ Preconditions.checkNotNull(path, "Path");
+
+ Savepoint savepoint;
+ synchronized (shutDownLock) {
+ savepoint = savepoints.get(path);
+ }
+
+ if (savepoint != null) {
+ return savepoint;
+ } else {
+ throw new IllegalArgumentException("Invalid path '" + path + "'.");
+ }
+ }
+
+ @Override
+ public void disposeSavepoint(String path, ClassLoader classLoader) throws Exception {
+ Preconditions.checkNotNull(path, "Path");
+ Preconditions.checkNotNull(classLoader, "Class loader");
+
+ Savepoint savepoint;
+ synchronized (shutDownLock) {
+ savepoint = savepoints.remove(path);
+ }
+
+ if (savepoint != null) {
+ savepoint.dispose(classLoader);
+ } else {
+ throw new IllegalArgumentException("Invalid path '" + path + "'.");
+ }
+ }
+
+ @Override
+ public void shutdown() throws Exception {
+ synchronized (shutDownLock) {
+ // This is problematic as the user code class loader is not
+ // available at this point.
+ for (Savepoint savepoint : savepoints.values()) {
+ try {
+ savepoint.dispose(ClassLoader.getSystemClassLoader());
+ } catch (Throwable t) {
+ LOG.warn("Failed to dispose savepoint " + savepoint.getCheckpointId(), t);
+ }
+ }
+
+ savepoints.clear();
+
+ // Remove shutdown hook to prevent resource leaks, unless this is
+ // invoked by the shutdown hook itself.
+ if (shutdownHook != null && shutdownHook != Thread.currentThread()) {
+ try {
+ Runtime.getRuntime().removeShutdownHook(shutdownHook);
+ } catch (IllegalStateException ignored) {
+ // Race, JVM is in shutdown already, we can safely ignore this
+ } catch (Throwable t) {
+ LOG.warn("Failed to unregister shut down hook.");
+ }
+ }
+
+ shutDown = true;
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/Savepoint.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/Savepoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/Savepoint.java
new file mode 100644
index 0000000..7823ab9
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/Savepoint.java
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.checkpoint.savepoint;
+
+import org.apache.flink.runtime.checkpoint.CheckpointIDCounter;
+import org.apache.flink.runtime.checkpoint.TaskState;
+
+import java.util.Collection;
+
+/**
+ * Savepoints are manually-triggered snapshots from which a program can be
+ * resumed on submission.
+ *
+ * <p>In order to allow changes to the savepoint format between Flink versions,
+ * we allow different savepoint implementations (see subclasses of this
+ * interface).
+ *
+ * <p>Savepoints are serialized via a {@link SavepointSerializer} and stored
+ * via a {@link SavepointStore}.
+ */
+public interface Savepoint {
+
+ /**
+ * Returns the savepoint version.
+ *
+ * <p>This version is independent of the Flink version, e.g. multiple Flink
+ * versions can work the same savepoint version.
+ *
+ * @return Savepoint version
+ */
+ int getVersion();
+
+ /**
+ * Returns the checkpoint ID of the savepoint.
+ *
+ * <p>Savepoints use the same mechanisms as regular checkpoints and are
+ * identified by a unique ID. This ID is used to reset the {@link CheckpointIDCounter}
+ * when restoring from a savepoint.
+ *
+ * @return Checkpoint ID of the savepoint
+ */
+ long getCheckpointId();
+
+ /**
+ * Returns the snapshotted task states.
+ *
+ * <p>These are used to restore the snapshot state.
+ *
+ * @return Snapshotted task states
+ */
+ Collection<TaskState> getTaskStates();
+
+ /**
+ * Disposes the savepoint.
+ *
+ * <p>The class loader is needed, because savepoints can currently point to
+ * arbitrary snapshot {@link org.apache.flink.runtime.state.StateHandle}
+ * instances, which need the user code class loader for deserialization.
+ *
+ * @param classLoader Class loader for disposal
+ */
+ void dispose(ClassLoader classLoader) throws Exception;
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointCoordinator.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointCoordinator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointCoordinator.java
new file mode 100644
index 0000000..5759e35
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointCoordinator.java
@@ -0,0 +1,387 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.checkpoint.savepoint;
+
+import akka.actor.ActorSystem;
+import akka.actor.Props;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.checkpoint.CompletedCheckpoint;
+import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
+import org.apache.flink.runtime.checkpoint.CheckpointIDCounter;
+import org.apache.flink.runtime.checkpoint.CompletedCheckpointStore;
+import org.apache.flink.runtime.checkpoint.SubtaskState;
+import org.apache.flink.runtime.checkpoint.TaskState;
+import org.apache.flink.runtime.checkpoint.stats.CheckpointStatsTracker;
+import org.apache.flink.runtime.execution.ExecutionState;
+import org.apache.flink.runtime.executiongraph.Execution;
+import org.apache.flink.runtime.executiongraph.ExecutionJobVertex;
+import org.apache.flink.runtime.executiongraph.ExecutionVertex;
+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.runtime.state.StateHandle;
+import org.apache.flink.util.SerializedValue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import scala.concurrent.Future;
+import scala.concurrent.Promise;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * The savepoint coordinator is a slightly modified variant of the regular
+ * checkpoint coordinator. Checkpoints are not triggered periodically, but
+ * manually. The actual checkpointing mechanism is the same as for periodic
+ * checkpoints, only the control flow is modified.
+ *
+ * <p>The savepoint coordinator is meant to be used as a separate coordinator
+ * instance. Otherwise, there can be unwanted queueing effects like discarding
+ * savepoints, because of in-progress periodic checkpoints.
+ *
+ * <p>The savepoint coordinator registers callbacks on the regular checkpoint
+ * life-cycle and manages a map of promises, which are completed/failed as soon
+ * as the trigged checkpoint is done.
+ *
+ * <p><strong>Important</strong>: it's necessary that both the periodic
+ * checkpoint coordinator and the savepoint coordinator <em>share</em> the same
+ * instance of the {@link CheckpointIDCounter} to ensure that all task managers
+ * see ascending checkpoints IDs.
+ */
+public class SavepointCoordinator extends CheckpointCoordinator {
+
+ private static final Logger LOG = LoggerFactory.getLogger(SavepointCoordinator.class);
+
+ /** Store for savepoints. */
+ private final SavepointStore savepointStore;
+
+ /** Mapping from checkpoint ID to promises for savepoints. */
+ private final Map<Long, Promise<String>> savepointPromises;
+
+ // TODO(uce) Temporary work around to restore initial state on
+ // failure during recovery. Will be superseded by FLINK-3397.
+ private volatile String savepointRestorePath;
+
+ public SavepointCoordinator(
+ JobID jobId,
+ long baseInterval,
+ long checkpointTimeout,
+ int numberKeyGroups,
+ ExecutionVertex[] tasksToTrigger,
+ ExecutionVertex[] tasksToWaitFor,
+ ExecutionVertex[] tasksToCommitTo,
+ ClassLoader userClassLoader,
+ CheckpointIDCounter checkpointIDCounter,
+ SavepointStore savepointStore,
+ CheckpointStatsTracker statsTracker) throws Exception {
+
+ super(jobId,
+ baseInterval,
+ checkpointTimeout,
+ 0L,
+ Integer.MAX_VALUE,
+ numberKeyGroups,
+ tasksToTrigger,
+ tasksToWaitFor,
+ tasksToCommitTo,
+ userClassLoader,
+ checkpointIDCounter,
+ IgnoreCheckpointsStore.INSTANCE,
+ RecoveryMode.STANDALONE,
+ statsTracker);
+
+ this.savepointStore = checkNotNull(savepointStore);
+ this.savepointPromises = new ConcurrentHashMap<>();
+ }
+
+ public String getSavepointRestorePath() {
+ return savepointRestorePath;
+ }
+
+ // ------------------------------------------------------------------------
+ // Savepoint trigger and reset
+ // ------------------------------------------------------------------------
+
+ /**
+ * Triggers a new savepoint using the current system time as the checkpoint timestamp.
+ */
+ public Future<String> triggerSavepoint(long timestamp) throws Exception {
+ final Promise<String> promise = new scala.concurrent.impl.Promise.DefaultPromise<>();
+
+ try {
+ // Get the checkpoint ID up front. If we fail to trigger the checkpoint,
+ // the ID will have changed, but this is OK as long as the checkpoint ID
+ // generates ascending IDs.
+ final long checkpointId = getAndIncrementCheckpointId();
+
+ if (checkpointId == -1) {
+ throw new IllegalStateException("Failed to get checkpoint Id");
+ }
+
+ LOG.info("Triggering savepoint with ID " + checkpointId);
+
+ // Important: make sure to add the promise to the map before calling
+ // any methods that might trigger callbacks, which require the promise.
+ // Otherwise, the might be race conditions.
+ if (savepointPromises.put(checkpointId, promise) == null) {
+ boolean success = false;
+
+ try {
+ // All good. The future will be completed as soon as the
+ // triggered checkpoint is done.
+ success = triggerCheckpoint(timestamp, checkpointId);
+ }
+ finally {
+ if (!success) {
+ savepointPromises.remove(checkpointId);
+ promise.failure(new Exception("Failed to trigger savepoint"));
+ }
+ }
+ }
+ else {
+ throw new IllegalStateException("Duplicate checkpoint ID");
+ }
+ }
+ catch (Throwable t) {
+ promise.failure(new Exception("Failed to trigger savepoint", t));
+ }
+
+ return promise.future();
+ }
+
+ /**
+ * Resets the state of {@link Execution} instances back to the state of a savepoint.
+ *
+ * <p>The execution vertices need to be in state {@link ExecutionState#CREATED} when calling
+ * this method. The operation might block. Make sure that calls don't block the job manager
+ * actor.
+ *
+ * @param tasks Tasks that will possibly be reset
+ * @param savepointPath The path of the savepoint to rollback to
+ * @throws IllegalStateException If coordinator is shut down
+ * @throws IllegalStateException If mismatch between program and savepoint state
+ * @throws Exception If savepoint store failure
+ */
+ public void restoreSavepoint(
+ Map<JobVertexID, ExecutionJobVertex> tasks,
+ String savepointPath) throws Exception {
+
+ checkNotNull(savepointPath, "Savepoint path");
+
+ synchronized (lock) {
+ if (isShutdown()) {
+ throw new IllegalStateException("CheckpointCoordinator is shut down");
+ }
+
+ LOG.info("Rolling back to savepoint '{}'.", savepointPath);
+
+ Savepoint savepoint = savepointStore.loadSavepoint(savepointPath);
+
+ long recoveryTimestamp = System.currentTimeMillis();
+
+ for (TaskState taskState : savepoint.getTaskStates()) {
+ ExecutionJobVertex executionJobVertex = tasks.get(taskState.getJobVertexID());
+
+ if (executionJobVertex != null) {
+ if (executionJobVertex.getParallelism() != taskState.getParallelism()) {
+ String msg = String.format("Failed to rollback to savepoint %s. " +
+ "Parallelism mismatch between savepoint state and new program. " +
+ "Cannot map operator %s with parallelism %d to new program with " +
+ "parallelism %d. This indicates that the program has been changed " +
+ "in a non-compatible way after the savepoint.",
+ savepoint,
+ taskState.getJobVertexID(),
+ taskState.getParallelism(),
+ executionJobVertex.getParallelism());
+
+ throw new IllegalStateException(msg);
+ }
+
+ List<Set<Integer>> keyGroupPartitions = createKeyGroupPartitions(
+ numberKeyGroups,
+ executionJobVertex.getParallelism());
+
+ for (int i = 0; i < executionJobVertex.getTaskVertices().length; i++) {
+ SubtaskState subtaskState = taskState.getState(i);
+ SerializedValue<StateHandle<?>> state = null;
+
+ if (subtaskState != null) {
+ state = subtaskState.getState();
+ }
+
+ Map<Integer, SerializedValue<StateHandle<?>>> kvStateForTaskMap = taskState
+ .getUnwrappedKvStates(keyGroupPartitions.get(i));
+
+ Execution currentExecutionAttempt = executionJobVertex
+ .getTaskVertices()[i]
+ .getCurrentExecutionAttempt();
+
+ currentExecutionAttempt.setInitialState(state, kvStateForTaskMap);
+ }
+ } else {
+ 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 in a " +
+ "non-compatible way after the savepoint.", savepointPath,
+ taskState.getJobVertexID());
+ throw new IllegalStateException(msg);
+ }
+ }
+
+ // Reset the checkpoint ID counter
+ long nextCheckpointId = savepoint.getCheckpointId() + 1;
+ checkpointIdCounter.start();
+ checkpointIdCounter.setCount(nextCheckpointId);
+ LOG.info("Reset the checkpoint ID to {}", nextCheckpointId);
+
+ if (savepointRestorePath == null) {
+ savepointRestorePath = savepointPath;
+ }
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // Checkpoint coordinator callbacks
+ // ------------------------------------------------------------------------
+
+ @Override
+ protected void onShutdown() {
+ // Fail all outstanding savepoint futures
+ for (Promise<String> promise : savepointPromises.values()) {
+ promise.failure(new Exception("Checkpoint coordinator shutdown"));
+ }
+ savepointPromises.clear();
+ }
+
+ @Override
+ protected void onCancelCheckpoint(long canceledCheckpointId) {
+ LOG.info("Cancelling savepoint with checkpoint ID " + canceledCheckpointId);
+ Promise<String> promise = savepointPromises.remove(canceledCheckpointId);
+
+ if (promise != null) {
+ promise.failure(new Exception("Savepoint expired before completing"));
+ }
+ }
+
+ @Override
+ protected void onFullyAcknowledgedCheckpoint(CompletedCheckpoint checkpoint) {
+ // Sanity check
+ Promise<String> promise = savepointPromises.remove(checkpoint.getCheckpointID());
+
+ if (promise == null) {
+ LOG.info("Pending savepoint with ID " + checkpoint.getCheckpointID() + " has been " +
+ "removed before receiving acknowledgment.");
+ }
+
+ // Sanity check
+ if (promise.isCompleted()) {
+ throw new IllegalStateException("Savepoint promise completed");
+ }
+
+ try {
+ Savepoint savepoint = new SavepointV0(
+ checkpoint.getCheckpointID(),
+ checkpoint.getTaskStates().values());
+ String path = savepointStore.storeSavepoint(savepoint);
+ promise.success(path);
+ }
+ catch (Exception e) {
+ LOG.warn("Failed to store savepoint.", e);
+ promise.failure(e);
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // Job status listener
+ // ------------------------------------------------------------------------
+
+ @Override
+ public ActorGateway createActivatorDeactivator(
+ ActorSystem actorSystem,
+ UUID leaderSessionID) {
+
+ synchronized (lock) {
+ if (isShutdown()) {
+ throw new IllegalArgumentException("Checkpoint coordinator is shut down");
+ }
+
+ if (getJobStatusListener() == null) {
+ Props props = Props.create(
+ SavepointCoordinatorDeActivator.class,
+ this,
+ leaderSessionID);
+
+ // wrap the ActorRef in a AkkaActorGateway to support message decoration
+ setJobStatusListener(new AkkaActorGateway(
+ actorSystem.actorOf(props),
+ leaderSessionID));
+ }
+
+ return getJobStatusListener();
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // Completed checkpoints
+ // ------------------------------------------------------------------------
+
+ private static class IgnoreCheckpointsStore implements CompletedCheckpointStore {
+
+ private static final CompletedCheckpointStore INSTANCE = new IgnoreCheckpointsStore();
+
+ @Override
+ public void recover() throws Exception {
+ }
+
+ @Override
+ public void addCheckpoint(CompletedCheckpoint checkpoint) throws Exception {
+ }
+
+ @Override
+ public CompletedCheckpoint getLatestCheckpoint() throws Exception {
+ return null;
+ }
+
+ @Override
+ public void shutdown() throws Exception {
+ }
+
+ @Override
+ public void suspend() throws Exception {
+ }
+
+ @Override
+ public List<CompletedCheckpoint> getAllCheckpoints() throws Exception {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public int getNumberOfRetainedCheckpoints() {
+ return 0;
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointCoordinatorDeActivator.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointCoordinatorDeActivator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointCoordinatorDeActivator.java
new file mode 100644
index 0000000..7bd5797
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointCoordinatorDeActivator.java
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.checkpoint.savepoint;
+
+import org.apache.flink.runtime.akka.FlinkUntypedActor;
+import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
+import org.apache.flink.runtime.jobgraph.JobStatus;
+import org.apache.flink.runtime.messages.ExecutionGraphMessages;
+import org.apache.flink.util.Preconditions;
+
+import java.util.UUID;
+
+/**
+ * This actor listens to changes in the JobStatus and deactivates the
+ * savepoint scheduler and discards all pending checkpoints.
+ */
+public class SavepointCoordinatorDeActivator extends FlinkUntypedActor {
+
+ private final CheckpointCoordinator coordinator;
+ private final UUID leaderSessionID;
+
+ public SavepointCoordinatorDeActivator(
+ SavepointCoordinator coordinator,
+ UUID leaderSessionID) {
+
+ LOG.info("Create SavepointCoordinatorDeActivator");
+
+ this.coordinator = Preconditions.checkNotNull(coordinator, "The checkpointCoordinator must not be null.");
+ this.leaderSessionID = leaderSessionID;
+ }
+
+ @Override
+ public void handleMessage(Object message) throws Exception {
+ if (message instanceof ExecutionGraphMessages.JobStatusChanged) {
+ JobStatus status = ((ExecutionGraphMessages.JobStatusChanged) message).newJobStatus();
+
+ if (status != JobStatus.RUNNING) {
+ // anything other than RUNNING should stop the trigger for now
+ coordinator.stopCheckpointScheduler();
+ }
+ }
+ }
+
+ @Override
+ public UUID getLeaderSessionID() {
+ return leaderSessionID;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointSerializer.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointSerializer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointSerializer.java
new file mode 100644
index 0000000..211209c
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointSerializer.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.checkpoint.savepoint;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+
+/**
+ * Serializer for {@link Savepoint} instances.
+ *
+ * <p>This serializer is used to read/write a savepoint via {@link FsSavepointStore}.
+ *
+ * <p>Version-specific serializers are accessed via the {@link SavepointSerializers} helper.
+ *
+ * @param <T> Savepoint type to serialize.
+ */
+public interface SavepointSerializer<T extends Savepoint> {
+
+ /**
+ * Serializes a savepoint to an output stream.
+ *
+ * @param savepoint Savepoint to serialize
+ * @param dos Output stream to serialize the savepoint to
+ * @throws IOException Serialization failures are forwarded
+ */
+ void serialize(T savepoint, DataOutputStream dos) throws IOException;
+
+ /**
+ * Deserializes a savepoint from an input stream.
+ *
+ * @param dis Input stream to deserialize savepoint from
+ * @return The deserialized savepoint
+ * @throws IOException Serialization failures are forwarded
+ */
+ T deserialize(DataInputStream dis) throws IOException;
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointSerializers.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointSerializers.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointSerializers.java
new file mode 100644
index 0000000..cc5c208
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointSerializers.java
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.checkpoint.savepoint;
+
+import org.apache.flink.util.Preconditions;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Helper to access {@link SavepointSerializer} for a specific savepoint version.
+ */
+public class SavepointSerializers {
+
+ private static final Map<Integer, SavepointSerializer<?>> SERIALIZERS = new HashMap<>(1);
+
+ static {
+ SERIALIZERS.put(SavepointV0.VERSION, SavepointV1Serializer.INSTANCE);
+ }
+
+ /**
+ * Returns the {@link SavepointSerializer} for the given savepoint.
+ *
+ * @param savepoint Savepoint to get serializer for
+ * @param <T> Type of savepoint
+ * @return Savepoint serializer for the savepoint
+ * @throws IllegalArgumentException If unknown savepoint version
+ */
+ @SuppressWarnings("unchecked")
+ public static <T extends Savepoint> SavepointSerializer<T> getSerializer(T savepoint) {
+ Preconditions.checkNotNull(savepoint, "Savepoint");
+ SavepointSerializer<T> serializer = (SavepointSerializer<T>) SERIALIZERS.get(savepoint.getVersion());
+ if (serializer != null) {
+ return serializer;
+ } else {
+ throw new IllegalArgumentException("Unknown savepoint version " + savepoint.getVersion() + ".");
+ }
+ }
+
+ /**
+ * Returns the {@link SavepointSerializer} for the given savepoint version.
+ *
+ * @param version Savepoint version to get serializer for
+ * @return Savepoint for the given version
+ * @throws IllegalArgumentException If unknown savepoint version
+ */
+ @SuppressWarnings("unchecked")
+ public static SavepointSerializer<?> getSerializer(int version) {
+ SavepointSerializer<?> serializer = SERIALIZERS.get(version);
+ if (serializer != null) {
+ return serializer;
+ } else {
+ throw new IllegalArgumentException("Unknown savepoint version " + version + ".");
+ }
+ }
+
+}
[2/3] flink git commit: [FLINK-4067] [runtime] Add savepoint headers
Posted by uc...@apache.org.
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointStore.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointStore.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointStore.java
new file mode 100644
index 0000000..71fcb34
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointStore.java
@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.checkpoint.savepoint;
+
+/**
+ * Savepoint store used to persist {@link Savepoint} instances.
+ *
+ * <p>The main implementation is the {@link FsSavepointStore}. We also have the
+ * {@link HeapSavepointStore} for historical reasons (introduced in Flink 1.0).
+ */
+public interface SavepointStore {
+
+ /**
+ * Stores the savepoint.
+ *
+ * @param savepoint Savepoint to be stored
+ * @param <T> Savepoint type
+ * @return Path of stored savepoint
+ * @throws Exception Failures during store are forwarded
+ */
+ <T extends Savepoint> String storeSavepoint(T savepoint) throws Exception;
+
+ /**
+ * Loads the savepoint at the specified path.
+ *
+ * @param path Path of savepoint to load
+ * @return The loaded savepoint
+ * @throws Exception Failures during load are forwared
+ */
+ Savepoint loadSavepoint(String path) throws Exception;
+
+ /**
+ * Disposes the savepoint at the specified path.
+ *
+ * <p>The class loader is needed, because savepoints can currently point to
+ * arbitrary snapshot {@link org.apache.flink.runtime.state.StateHandle}
+ * instances, which need the user code class loader for deserialization.
+ *
+ * @param path Path of savepoint to dispose
+ * @param classLoader Class loader for disposal
+ * @throws Exception Failures during diposal are forwarded
+ */
+ void disposeSavepoint(String path, ClassLoader classLoader) throws Exception;
+
+ /**
+ * Shut downs the savepoint store.
+ *
+ * <p>Only necessary for implementations where the savepoint life-cycle is
+ * bound to the cluster life-cycle.
+ *
+ * @throws Exception Failures during shut down are forwarded
+ */
+ void shutdown() throws Exception;
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointStoreFactory.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointStoreFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointStoreFactory.java
new file mode 100644
index 0000000..d2bf40e
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointStoreFactory.java
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.checkpoint.savepoint;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.IllegalConfigurationException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+/**
+ * Factory for {@link SavepointStore} instances.
+ */
+public class SavepointStoreFactory {
+
+ public static final String SAVEPOINT_BACKEND_KEY = "savepoints.state.backend";
+ public static final String SAVEPOINT_DIRECTORY_KEY = "savepoints.state.backend.fs.dir";
+ public static final String DEFAULT_SAVEPOINT_BACKEND = "jobmanager";
+
+ public static final Logger LOG = LoggerFactory.getLogger(SavepointStoreFactory.class);
+
+ /**
+ * Creates a {@link SavepointStore} from the specified Configuration.
+ *
+ * <p>You can configure a savepoint-specific backend for the savepoints. If
+ * you don't configure anything, the regular checkpoint backend will be
+ * used.
+ *
+ * <p>The default and fallback backend is the job manager, which loses the
+ * savepoint after the job manager shuts down.
+ *
+ * @param config The configuration to parse the savepoint backend configuration.
+ * @return A savepoint store.
+ */
+ public static SavepointStore createFromConfig(Configuration config) throws Exception {
+
+ // Try a the savepoint-specific configuration first.
+ String savepointBackend = config.getString(SAVEPOINT_BACKEND_KEY, DEFAULT_SAVEPOINT_BACKEND);
+
+ if (savepointBackend == null) {
+ LOG.info("No savepoint state backend configured. " +
+ "Using job manager savepoint state backend.");
+ return createJobManagerSavepointStore();
+ } else if (savepointBackend.equals("jobmanager")) {
+ LOG.info("Using job manager savepoint state backend.");
+ return createJobManagerSavepointStore();
+ } else if (savepointBackend.equals("filesystem")) {
+ String rootPath = config.getString(SAVEPOINT_DIRECTORY_KEY, null);
+
+ if (rootPath == null) {
+ throw new IllegalConfigurationException("Using filesystem as savepoint state backend, " +
+ "but did not specify directory. Please set the " +
+ "following configuration key: '" + SAVEPOINT_DIRECTORY_KEY +
+ "' (e.g. " + SAVEPOINT_DIRECTORY_KEY + ": hdfs:///flink/savepoints/). " +
+ "Falling back to job manager savepoint backend.");
+ } else {
+ LOG.info("Using filesystem savepoint backend (root path: {}).", rootPath);
+
+ return createFileSystemSavepointStore(rootPath);
+ }
+ } else {
+ throw new IllegalConfigurationException("Unexpected savepoint backend " +
+ "configuration '" + savepointBackend + "'. " +
+ "Falling back to job manager savepoint state backend.");
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // Savepoint stores
+ // ------------------------------------------------------------------------
+
+ private static SavepointStore createJobManagerSavepointStore() {
+ return new HeapSavepointStore();
+ }
+
+ private static SavepointStore createFileSystemSavepointStore(String rootPath) throws IOException {
+ return new FsSavepointStore(rootPath, "savepoint-");
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointV0.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointV0.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointV0.java
new file mode 100644
index 0000000..9fd950d
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointV0.java
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.checkpoint.savepoint;
+
+import org.apache.flink.runtime.checkpoint.TaskState;
+import org.apache.flink.util.Preconditions;
+
+import java.util.Collection;
+
+/**
+ * Savepoint version 0.
+ *
+ * <p>This format was introduced with Flink 1.1.0.
+ */
+public class SavepointV0 implements Savepoint {
+
+ /** The savepoint version. */
+ public static final int VERSION = 0;
+
+ /** The checkpoint ID */
+ private final long checkpointId;
+
+ /** The task states */
+ private final Collection<TaskState> taskStates;
+
+ SavepointV0(long checkpointId, Collection<TaskState> taskStates) {
+ this.checkpointId = checkpointId;
+ this.taskStates = Preconditions.checkNotNull(taskStates, "Task States");
+ }
+
+ @Override
+ public int getVersion() {
+ return VERSION;
+ }
+
+ @Override
+ public long getCheckpointId() {
+ return checkpointId;
+ }
+
+ @Override
+ public Collection<TaskState> getTaskStates() {
+ return taskStates;
+ }
+
+ @Override
+ public void dispose(ClassLoader classLoader) throws Exception {
+ Preconditions.checkNotNull(classLoader, "Class loader");
+ for (TaskState taskState : taskStates) {
+ taskState.discard(classLoader);
+ }
+ taskStates.clear();
+ }
+
+ @Override
+ public String toString() {
+ return "Savepoint(version=" + VERSION + ")";
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ SavepointV0 that = (SavepointV0) o;
+ return checkpointId == that.checkpointId && getTaskStates().equals(that.getTaskStates());
+ }
+
+ @Override
+ public int hashCode() {
+ int result = (int) (checkpointId ^ (checkpointId >>> 32));
+ result = 31 * result + taskStates.hashCode();
+ return result;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointV1Serializer.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointV1Serializer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointV1Serializer.java
new file mode 100644
index 0000000..8de29a6
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointV1Serializer.java
@@ -0,0 +1,186 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.checkpoint.savepoint;
+
+import org.apache.flink.runtime.checkpoint.KeyGroupState;
+import org.apache.flink.runtime.checkpoint.SubtaskState;
+import org.apache.flink.runtime.checkpoint.TaskState;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.state.StateHandle;
+import org.apache.flink.util.SerializedValue;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+/**
+ * Serializer for {@link SavepointV0} instances.
+ *
+ * <p>In contrast to previous savepoint versions, this serializer makes sure
+ * that no default Java serialization is used for serialization. Therefore, we
+ * don't rely on any involved Java classes to stay the same.
+ */
+class SavepointV1Serializer implements SavepointSerializer<SavepointV0> {
+
+ public static final SavepointV1Serializer INSTANCE = new SavepointV1Serializer();
+
+ private SavepointV1Serializer() {
+ }
+
+ @Override
+ public void serialize(SavepointV0 savepoint, DataOutputStream dos) throws IOException {
+ dos.writeLong(savepoint.getCheckpointId());
+
+ Collection<TaskState> taskStates = savepoint.getTaskStates();
+ dos.writeInt(taskStates.size());
+
+ for (TaskState taskState : savepoint.getTaskStates()) {
+ // Vertex ID
+ dos.writeLong(taskState.getJobVertexID().getLowerPart());
+ dos.writeLong(taskState.getJobVertexID().getUpperPart());
+
+ // Parallelism
+ int parallelism = taskState.getParallelism();
+ dos.writeInt(parallelism);
+
+ // Sub task states
+ dos.writeInt(taskState.getNumberCollectedStates());
+
+ for (int i = 0; i < parallelism; i++) {
+ SubtaskState subtaskState = taskState.getState(i);
+
+ if (subtaskState != null) {
+ dos.writeInt(i);
+
+ SerializedValue<?> serializedValue = subtaskState.getState();
+ if (serializedValue == null) {
+ dos.writeInt(-1); // null
+ } else {
+ byte[] serialized = serializedValue.getByteArray();
+ dos.writeInt(serialized.length);
+ dos.write(serialized, 0, serialized.length);
+ }
+
+ dos.writeLong(subtaskState.getStateSize());
+ dos.writeLong(subtaskState.getDuration());
+ }
+ }
+
+ // Key group states
+ dos.writeInt(taskState.getNumberCollectedKvStates());
+
+ for (int i = 0; i < parallelism; i++) {
+ KeyGroupState keyGroupState = taskState.getKvState(i);
+
+ if (keyGroupState != null) {
+ dos.write(i);
+
+ SerializedValue<?> serializedValue = keyGroupState.getKeyGroupState();
+ if (serializedValue == null) {
+ dos.writeInt(-1); // null
+ } else {
+ byte[] serialized = serializedValue.getByteArray();
+ dos.writeInt(serialized.length);
+ dos.write(serialized, 0, serialized.length);
+ }
+
+ dos.writeLong(keyGroupState.getStateSize());
+ dos.writeLong(keyGroupState.getDuration());
+ }
+ }
+ }
+ }
+
+ @Override
+ public SavepointV0 deserialize(DataInputStream dis) throws IOException {
+ long checkpointId = dis.readLong();
+
+ // Task states
+ int numTaskStates = dis.readInt();
+ List<TaskState> taskStates = new ArrayList<>(numTaskStates);
+
+ for (int i = 0; i < numTaskStates; i++) {
+ JobVertexID jobVertexId = new JobVertexID(dis.readLong(), dis.readLong());
+ int parallelism = dis.readInt();
+
+ // Add task state
+ TaskState taskState = new TaskState(jobVertexId, parallelism);
+ taskStates.add(taskState);
+
+ // Sub task states
+ int numSubTaskStates = dis.readInt();
+ for (int j = 0; j < numSubTaskStates; j++) {
+ int subtaskIndex = dis.readInt();
+
+ int length = dis.readInt();
+
+ SerializedValue<StateHandle<?>> serializedValue;
+ if (length == -1) {
+ serializedValue = new SerializedValue<>(null);
+ } else {
+ byte[] serializedData = new byte[length];
+ dis.read(serializedData, 0, length);
+ serializedValue = SerializedValue.fromBytes(serializedData);
+ }
+
+ long stateSize = dis.readLong();
+ long duration = dis.readLong();
+
+ SubtaskState subtaskState = new SubtaskState(
+ serializedValue,
+ stateSize,
+ duration);
+
+ taskState.putState(subtaskIndex, subtaskState);
+ }
+
+ // Key group states
+ int numKvStates = dis.readInt();
+ for (int j = 0; j < numKvStates; j++) {
+ int keyGroupIndex = dis.readInt();
+
+ int length = dis.readInt();
+
+ SerializedValue<StateHandle<?>> serializedValue;
+ if (length == -1) {
+ serializedValue = new SerializedValue<>(null);
+ } else {
+ byte[] serializedData = new byte[length];
+ dis.read(serializedData, 0, length);
+ serializedValue = SerializedValue.fromBytes(serializedData);
+ }
+
+ long stateSize = dis.readLong();
+ long duration = dis.readLong();
+
+ KeyGroupState keyGroupState = new KeyGroupState(
+ serializedValue,
+ stateSize,
+ duration);
+
+ taskState.putKvState(keyGroupIndex, keyGroupState);
+ }
+ }
+
+ return new SavepointV0(checkpointId, taskStates);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/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 d405b88..4229105 100755
--- 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
@@ -35,10 +35,9 @@ import org.apache.flink.runtime.accumulators.StringifiedAccumulatorResult;
import org.apache.flink.runtime.blob.BlobKey;
import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
import org.apache.flink.runtime.checkpoint.CheckpointIDCounter;
-import org.apache.flink.runtime.checkpoint.CompletedCheckpoint;
import org.apache.flink.runtime.checkpoint.CompletedCheckpointStore;
-import org.apache.flink.runtime.checkpoint.SavepointCoordinator;
-import org.apache.flink.runtime.checkpoint.StateStore;
+import org.apache.flink.runtime.checkpoint.savepoint.SavepointCoordinator;
+import org.apache.flink.runtime.checkpoint.savepoint.SavepointStore;
import org.apache.flink.runtime.checkpoint.stats.CheckpointStatsTracker;
import org.apache.flink.runtime.execution.ExecutionState;
import org.apache.flink.runtime.execution.SuppressRestartsException;
@@ -361,9 +360,9 @@ public class ExecutionGraph implements Serializable {
ActorSystem actorSystem,
UUID leaderSessionID,
CheckpointIDCounter checkpointIDCounter,
- CompletedCheckpointStore completedCheckpointStore,
+ CompletedCheckpointStore checkpointStore,
RecoveryMode recoveryMode,
- StateStore<CompletedCheckpoint> savepointStore,
+ SavepointStore savepointStore,
CheckpointStatsTracker statsTracker) throws Exception {
// simple sanity checks
@@ -396,7 +395,7 @@ public class ExecutionGraph implements Serializable {
tasksToCommitTo,
userClassLoader,
checkpointIDCounter,
- completedCheckpointStore,
+ checkpointStore,
recoveryMode,
checkpointStatsTracker);
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobVertexID.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobVertexID.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobVertexID.java
index 868c124..514aabc 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobVertexID.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobVertexID.java
@@ -32,7 +32,11 @@ public class JobVertexID extends AbstractID {
public JobVertexID() {
super();
}
-
+
+ public JobVertexID(long lowerPart, long upperPart) {
+ super(lowerPart, upperPart);
+ }
+
public JobVertexID(byte[] bytes) {
super(bytes);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/main/java/org/apache/flink/runtime/util/ZooKeeperUtils.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/util/ZooKeeperUtils.java b/flink-runtime/src/main/java/org/apache/flink/runtime/util/ZooKeeperUtils.java
index 5d3a5c4..3986fed 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/util/ZooKeeperUtils.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/util/ZooKeeperUtils.java
@@ -25,8 +25,8 @@ import org.apache.flink.api.common.JobID;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.IllegalConfigurationException;
-import org.apache.flink.runtime.checkpoint.CompletedCheckpoint;
import org.apache.flink.runtime.checkpoint.CompletedCheckpointStore;
+import org.apache.flink.runtime.checkpoint.CompletedCheckpoint;
import org.apache.flink.runtime.checkpoint.ZooKeeperCheckpointIDCounter;
import org.apache.flink.runtime.checkpoint.ZooKeeperCompletedCheckpointStore;
import org.apache.flink.runtime.jobmanager.RecoveryMode;
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/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 ccbd263..ef52381 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
@@ -41,6 +41,7 @@ import org.apache.flink.runtime.accumulators.AccumulatorSnapshot
import org.apache.flink.runtime.akka.{AkkaUtils, ListeningBehaviour}
import org.apache.flink.runtime.blob.BlobServer
import org.apache.flink.runtime.checkpoint._
+import org.apache.flink.runtime.checkpoint.savepoint.{SavepointStoreFactory, SavepointStore}
import org.apache.flink.runtime.checkpoint.stats.{CheckpointStatsTracker, SimpleCheckpointStatsTracker, DisabledCheckpointStatsTracker}
import org.apache.flink.runtime.client._
import org.apache.flink.runtime.execution.SuppressRestartsException
@@ -212,14 +213,6 @@ class JobManager(
throw new RuntimeException("Could not start the checkpoint recovery service.", e)
}
- try {
- savepointStore.start()
- } catch {
- case e: Exception =>
- log.error("Could not start the savepoint store.", e)
- throw new RuntimeException("Could not start the savepoint store store.", e)
- }
-
jobManagerMetricGroup match {
case Some(group) =>
instantiateMetrics(group)
@@ -249,10 +242,10 @@ class JobManager(
}
try {
- savepointStore.stop()
+ savepointStore.shutdown()
} catch {
case e: Exception =>
- log.error("Could not stop the savepoint store.", e)
+ log.error("Could not shut down savepoint store.", e)
throw new RuntimeException("Could not stop the savepoint store store.", e)
}
@@ -751,10 +744,6 @@ class JobManager(
try {
log.info(s"Disposing savepoint at '$savepointPath'.")
- val savepoint = savepointStore.getState(savepointPath)
-
- log.debug(s"$savepoint")
-
if (blobKeys.isDefined) {
// We don't need a real ID here for the library cache manager
val jid = new JobID()
@@ -764,18 +753,15 @@ class JobManager(
val classLoader = libraryCacheManager.getClassLoader(jid)
// Discard with user code loader
- savepoint.discard(classLoader)
+ savepointStore.disposeSavepoint(savepointPath, classLoader)
} finally {
libraryCacheManager.unregisterJob(jid)
}
} else {
// Discard with system class loader
- savepoint.discard(getClass.getClassLoader)
+ savepointStore.disposeSavepoint(savepointPath, getClass.getClassLoader)
}
- // Dispose the savepoint
- savepointStore.disposeState(savepointPath)
-
senderRef ! DisposeSavepointSuccess
} catch {
case t: Throwable =>
@@ -1237,7 +1223,7 @@ class JobManager(
snapshotSettings.getVerticesToConfirm().asScala.map(idToVertex).asJava
val completedCheckpoints = checkpointRecoveryFactory
- .createCompletedCheckpoints(jobId, userCodeLoader)
+ .createCheckpointStore(jobId, userCodeLoader)
val checkpointIdCounter = checkpointRecoveryFactory.createCheckpointIDCounter(jobId)
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/main/scala/org/apache/flink/runtime/testingUtils/TestingJobManager.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/testingUtils/TestingJobManager.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/testingUtils/TestingJobManager.scala
index 04689c6..bebcf7b 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/testingUtils/TestingJobManager.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/testingUtils/TestingJobManager.scala
@@ -22,7 +22,8 @@ import akka.actor.ActorRef
import org.apache.flink.configuration.Configuration
import org.apache.flink.metrics.MetricRegistry
-import org.apache.flink.runtime.checkpoint.{SavepointStore, CheckpointRecoveryFactory}
+import org.apache.flink.runtime.checkpoint.CheckpointRecoveryFactory
+import org.apache.flink.runtime.checkpoint.savepoint.SavepointStore
import org.apache.flink.runtime.execution.librarycache.BlobLibraryCacheManager
import org.apache.flink.runtime.executiongraph.restart.RestartStrategyFactory
import org.apache.flink.runtime.instance.InstanceManager
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/main/scala/org/apache/flink/runtime/testingUtils/TestingJobManagerLike.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/testingUtils/TestingJobManagerLike.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/testingUtils/TestingJobManagerLike.scala
index 5ccebe2..2fe830f 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/testingUtils/TestingJobManagerLike.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/testingUtils/TestingJobManagerLike.scala
@@ -293,7 +293,7 @@ trait TestingJobManagerLike extends FlinkActor {
case RequestSavepoint(savepointPath) =>
try {
- val savepoint = savepointStore.getState(savepointPath)
+ val savepoint = savepointStore.loadSavepoint(savepointPath)
sender ! ResponseSavepoint(savepoint)
}
catch {
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/main/scala/org/apache/flink/runtime/testingUtils/TestingJobManagerMessages.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/testingUtils/TestingJobManagerMessages.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/testingUtils/TestingJobManagerMessages.scala
index ec513e3..a411c8b 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/testingUtils/TestingJobManagerMessages.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/testingUtils/TestingJobManagerMessages.scala
@@ -28,6 +28,7 @@ import org.apache.flink.runtime.checkpoint.CompletedCheckpoint
import org.apache.flink.runtime.executiongraph.{ExecutionAttemptID, ExecutionGraph}
import org.apache.flink.runtime.instance.ActorGateway
import org.apache.flink.runtime.jobgraph.JobStatus
+import org.apache.flink.runtime.checkpoint.savepoint.Savepoint
object TestingJobManagerMessages {
@@ -107,7 +108,7 @@ object TestingJobManagerMessages {
*
* @param savepoint The requested savepoint or null if none available.
*/
- case class ResponseSavepoint(savepoint: CompletedCheckpoint)
+ case class ResponseSavepoint(savepoint: Savepoint)
def getNotifyWhenLeader(): AnyRef = NotifyWhenLeader
def getDisablePostStop(): AnyRef = DisablePostStop
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/AbstractStateStoreTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/AbstractStateStoreTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/AbstractStateStoreTest.java
deleted file mode 100644
index f9f844c..0000000
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/AbstractStateStoreTest.java
+++ /dev/null
@@ -1,63 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.checkpoint;
-
-import org.junit.Test;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-
-/**
- * Basic {@link StateStore} behaviour test.
- */
-public abstract class AbstractStateStoreTest {
-
- abstract StateStore<Integer> createStateStore() throws Exception;
-
- abstract boolean verifyDiscarded(StateStore<Integer> stateStore, String path);
-
- @Test
- public void testSimplePutGetDiscardSequence() throws Exception {
- StateStore<Integer> stateStore = createStateStore();
-
- // Put
- Integer expectedState = 211155;
- String path = stateStore.putState(expectedState);
-
- // Get
- Integer actualState = stateStore.getState(path);
- assertEquals(expectedState, actualState);
-
- // Dispose
- stateStore.disposeState(path);
- assertTrue(verifyDiscarded(stateStore, path));
- }
-
- @Test(expected = IllegalArgumentException.class)
- public void testGetStateInvalidPathThrowsException() throws Exception {
- StateStore<Integer> stateStore = createStateStore();
- stateStore.getState("testGetStateInvalidPathThrowsException");
- }
-
- @Test(expected = IllegalArgumentException.class)
- public void testDisposeStateInvalidPathThrowsException() throws Exception {
- StateStore<Integer> stateStore = createStateStore();
- stateStore.getState("testDisposeStateInvalidPathThrowsException");
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStoreTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStoreTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStoreTest.java
index 079c341..84d809a 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStoreTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStoreTest.java
@@ -71,7 +71,7 @@ public abstract class CompletedCheckpointStoreTest extends TestLogger {
assertEquals(0, checkpoints.getNumberOfRetainedCheckpoints());
assertEquals(0, checkpoints.getAllCheckpoints().size());
- TestCheckpoint[] expected = new TestCheckpoint[] {
+ TestCompletedCheckpoint[] expected = new TestCompletedCheckpoint[] {
createCheckpoint(0), createCheckpoint(1) };
// Add and get latest
@@ -92,7 +92,7 @@ public abstract class CompletedCheckpointStoreTest extends TestLogger {
public void testAddCheckpointMoreThanMaxRetained() throws Exception {
CompletedCheckpointStore checkpoints = createCompletedCheckpoints(1, userClassLoader);
- TestCheckpoint[] expected = new TestCheckpoint[] {
+ TestCompletedCheckpoint[] expected = new TestCompletedCheckpoint[] {
createCheckpoint(0), createCheckpoint(1),
createCheckpoint(2), createCheckpoint(3)
};
@@ -137,12 +137,12 @@ public abstract class CompletedCheckpointStoreTest extends TestLogger {
public void testGetAllCheckpoints() throws Exception {
CompletedCheckpointStore checkpoints = createCompletedCheckpoints(4, userClassLoader);
- TestCheckpoint[] expected = new TestCheckpoint[] {
+ TestCompletedCheckpoint[] expected = new TestCompletedCheckpoint[] {
createCheckpoint(0), createCheckpoint(1),
createCheckpoint(2), createCheckpoint(3)
};
- for (TestCheckpoint checkpoint : expected) {
+ for (TestCompletedCheckpoint checkpoint : expected) {
checkpoints.addCheckpoint(checkpoint);
}
@@ -162,12 +162,12 @@ public abstract class CompletedCheckpointStoreTest extends TestLogger {
public void testDiscardAllCheckpoints() throws Exception {
CompletedCheckpointStore checkpoints = createCompletedCheckpoints(4, userClassLoader);
- TestCheckpoint[] expected = new TestCheckpoint[] {
+ TestCompletedCheckpoint[] expected = new TestCompletedCheckpoint[] {
createCheckpoint(0), createCheckpoint(1),
createCheckpoint(2), createCheckpoint(3)
};
- for (TestCheckpoint checkpoint : expected) {
+ for (TestCompletedCheckpoint checkpoint : expected) {
checkpoints.addCheckpoint(checkpoint);
}
@@ -179,7 +179,7 @@ public abstract class CompletedCheckpointStoreTest extends TestLogger {
assertEquals(0, checkpoints.getNumberOfRetainedCheckpoints());
// All have been discarded
- for (TestCheckpoint checkpoint : expected) {
+ for (TestCompletedCheckpoint checkpoint : expected) {
// The ZooKeeper implementation discards asynchronously
checkpoint.awaitDiscard();
assertTrue(checkpoint.isDiscarded());
@@ -189,11 +189,11 @@ public abstract class CompletedCheckpointStoreTest extends TestLogger {
// ---------------------------------------------------------------------------------------------
- protected TestCheckpoint createCheckpoint(int id) throws IOException {
+ protected TestCompletedCheckpoint createCheckpoint(int id) throws IOException {
return createCheckpoint(id, 4);
}
- protected TestCheckpoint createCheckpoint(int id, int numberOfStates)
+ protected TestCompletedCheckpoint createCheckpoint(int id, int numberOfStates)
throws IOException {
JobVertexID jvid = new JobVertexID();
@@ -209,7 +209,7 @@ public abstract class CompletedCheckpointStoreTest extends TestLogger {
taskState.putState(i, new SubtaskState(stateHandle, 0, 0));
}
- return new TestCheckpoint(new JobID(), id, 0, taskGroupStates);
+ return new TestCompletedCheckpoint(new JobID(), id, 0, taskGroupStates);
}
private void verifyCheckpoint(CompletedCheckpoint expected, CompletedCheckpoint actual) {
@@ -221,7 +221,7 @@ public abstract class CompletedCheckpointStoreTest extends TestLogger {
* used when discarding. Spying on a regular {@link CompletedCheckpoint} instance with
* Mockito doesn't work, because it it breaks serializability.
*/
- protected static class TestCheckpoint extends CompletedCheckpoint {
+ protected static class TestCompletedCheckpoint extends CompletedCheckpoint {
private static final long serialVersionUID = 4211419809665983026L;
@@ -232,7 +232,7 @@ public abstract class CompletedCheckpointStoreTest extends TestLogger {
private transient ClassLoader discardClassLoader;
- public TestCheckpoint(
+ public TestCompletedCheckpoint(
JobID jobId,
long checkpointId,
long timestamp,
@@ -274,7 +274,7 @@ public abstract class CompletedCheckpointStoreTest extends TestLogger {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
- TestCheckpoint that = (TestCheckpoint) o;
+ TestCompletedCheckpoint that = (TestCompletedCheckpoint) o;
return getJobId().equals(that.getJobId())
&& getCheckpointID() == that.getCheckpointID();
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ExecutionGraphCheckpointCoordinatorTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ExecutionGraphCheckpointCoordinatorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ExecutionGraphCheckpointCoordinatorTest.java
index b9ad1bd..0f2c2b2 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ExecutionGraphCheckpointCoordinatorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ExecutionGraphCheckpointCoordinatorTest.java
@@ -26,6 +26,8 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.blob.BlobKey;
+import org.apache.flink.runtime.checkpoint.savepoint.HeapSavepointStore;
+import org.apache.flink.runtime.checkpoint.savepoint.SavepointCoordinator;
import org.apache.flink.runtime.checkpoint.stats.DisabledCheckpointStatsTracker;
import org.apache.flink.runtime.executiongraph.ExecutionGraph;
import org.apache.flink.runtime.executiongraph.ExecutionJobVertex;
@@ -148,7 +150,7 @@ public class ExecutionGraphCheckpointCoordinatorTest {
counter,
store,
RecoveryMode.STANDALONE,
- new HeapStateStore<CompletedCheckpoint>(),
+ new HeapSavepointStore(),
new DisabledCheckpointStatsTracker());
JobVertex jobVertex = new JobVertex("MockVertex");
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/FileSystemStateStoreTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/FileSystemStateStoreTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/FileSystemStateStoreTest.java
deleted file mode 100644
index 7779f61..0000000
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/FileSystemStateStoreTest.java
+++ /dev/null
@@ -1,68 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.checkpoint;
-
-import org.apache.flink.core.fs.Path;
-import org.apache.flink.runtime.testutils.CommonTestUtils;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
-
-import java.io.File;
-import java.util.Arrays;
-
-import static org.junit.Assert.assertFalse;
-
-/**
- * Basic {@link FileSystemStateStore} behaviour test.
- */
-public class FileSystemStateStoreTest extends AbstractStateStoreTest {
-
- private static File rootPath;
-
- @BeforeClass
- public static void setUp() throws Exception {
- rootPath = CommonTestUtils.createTempDirectory();
- }
-
- @AfterClass
- public static void tearDown() throws Exception {
- if (rootPath != null) {
- try {
- String errMsg = "Root path " + rootPath.getPath() + " not cleaned up. "
- + Arrays.toString(rootPath.listFiles());
- assertFalse(errMsg, rootPath.exists());
- }
- finally {
- org.apache.commons.io.FileUtils.deleteDirectory(rootPath);
- }
- }
- }
-
- @Override
- StateStore<Integer> createStateStore() throws Exception {
- return new FileSystemStateStore<>(new Path(rootPath.toURI()), "test_savepoint-");
- }
-
- @Override
- @SuppressWarnings("unchecked")
- boolean verifyDiscarded(StateStore<Integer> stateStore, String path) {
- return true;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/HeapStateStoreTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/HeapStateStoreTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/HeapStateStoreTest.java
deleted file mode 100644
index 86b8fde..0000000
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/HeapStateStoreTest.java
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.checkpoint;
-
-import org.mockito.internal.util.reflection.Whitebox;
-
-import java.util.Map;
-
-/**
- * Basic {@link HeapStateStore} behaviour test.
- */
-public class HeapStateStoreTest extends AbstractStateStoreTest {
-
- @Override
- StateStore<Integer> createStateStore() {
- return new HeapStateStore<>();
- }
-
- @Override
- @SuppressWarnings("unchecked")
- boolean verifyDiscarded(StateStore<Integer> stateStore, String path) {
- Map<Integer, Integer> stateMap = (Map<Integer, Integer>) Whitebox
- .getInternalState(stateStore, "stateMap");
-
- return !stateMap.containsKey(path);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/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
deleted file mode 100644
index b103793..0000000
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/SavepointCoordinatorTest.java
+++ /dev/null
@@ -1,1140 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.checkpoint;
-
-import org.apache.commons.io.FileUtils;
-import org.apache.flink.api.common.JobID;
-import org.apache.flink.runtime.checkpoint.stats.DisabledCheckpointStatsTracker;
-import org.apache.flink.runtime.execution.ExecutionState;
-import org.apache.flink.runtime.executiongraph.Execution;
-import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
-import org.apache.flink.runtime.executiongraph.ExecutionJobVertex;
-import org.apache.flink.runtime.executiongraph.ExecutionVertex;
-import org.apache.flink.runtime.jobgraph.JobVertexID;
-import org.apache.flink.runtime.jobmanager.RecoveryMode;
-import org.apache.flink.runtime.messages.checkpoint.AcknowledgeCheckpoint;
-import org.apache.flink.runtime.messages.checkpoint.DeclineCheckpoint;
-import org.apache.flink.runtime.messages.checkpoint.NotifyCheckpointComplete;
-import org.apache.flink.runtime.messages.checkpoint.TriggerCheckpoint;
-import org.apache.flink.runtime.state.LocalStateHandle;
-import org.apache.flink.runtime.state.StateHandle;
-import org.apache.flink.runtime.testutils.CommonTestUtils;
-import org.apache.flink.util.SerializedValue;
-import org.apache.flink.util.TestLogger;
-import org.junit.Test;
-import scala.concurrent.Await;
-import scala.concurrent.Future;
-import scala.concurrent.Promise;
-import scala.concurrent.duration.Deadline;
-import scala.concurrent.duration.FiniteDuration;
-
-import java.io.File;
-import java.io.IOException;
-import java.io.Serializable;
-import java.lang.reflect.Field;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNotEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
-import static org.mockito.Matchers.any;
-import static org.mockito.Matchers.anyString;
-import static org.mockito.Matchers.eq;
-import static org.mockito.Mockito.doThrow;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.spy;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-
-/**
- * Tests for the savepoint coordinator.
- */
-public class SavepointCoordinatorTest extends TestLogger {
-
- // ------------------------------------------------------------------------
- // Trigger and acknowledge
- // ------------------------------------------------------------------------
-
- /**
- * Simple trigger-acknowledge test for a single savepoint.
- */
- @Test
- public void testSimpleTriggerSavepoint() throws Exception {
- JobID jobId = new JobID();
- long checkpointTimeout = 60 * 1000;
- long timestamp = 1272635;
- ExecutionVertex[] vertices = new ExecutionVertex[] {
- mockExecutionVertex(jobId),
- mockExecutionVertex(jobId) };
- MockCheckpointIdCounter checkpointIdCounter = new MockCheckpointIdCounter();
- HeapStateStore<CompletedCheckpoint> savepointStore = new HeapStateStore<>();
-
- SavepointCoordinator coordinator = createSavepointCoordinator(
- jobId,
- checkpointTimeout,
- vertices,
- vertices,
- vertices,
- checkpointIdCounter,
- savepointStore);
-
- // Trigger the savepoint
- Future<String> savepointPathFuture = coordinator.triggerSavepoint(timestamp);
- assertFalse(savepointPathFuture.isCompleted());
-
- long checkpointId = checkpointIdCounter.getLastReturnedCount();
- assertEquals(0, checkpointId);
-
- // Verify send trigger messages
- for (ExecutionVertex vertex : vertices) {
- verifyTriggerCheckpoint(vertex, checkpointId, timestamp);
- }
-
- PendingCheckpoint pendingCheckpoint = coordinator.getPendingCheckpoints()
- .get(checkpointId);
-
- verifyPendingCheckpoint(pendingCheckpoint, jobId, checkpointId,
- timestamp, 0, 2, 0, false, false);
-
- // Acknowledge tasks
- for (ExecutionVertex vertex : vertices) {
- coordinator.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(
- jobId, vertex.getCurrentExecutionAttempt().getAttemptId(),
- checkpointId, createSerializedStateHandle(vertex), 0));
- }
-
- // The pending checkpoint is completed
- assertTrue(pendingCheckpoint.isDiscarded());
- assertEquals(0, coordinator.getSuccessfulCheckpoints().size());
-
- // Verify send notify complete messages
- for (ExecutionVertex vertex : vertices) {
- verifyNotifyCheckpointComplete(vertex, checkpointId, timestamp);
- }
-
- // Verify that the future has been completed
- assertTrue(savepointPathFuture.isCompleted());
- String savepointPath = Await.result(savepointPathFuture, FiniteDuration.Zero());
-
- // Verify the savepoint
- CompletedCheckpoint savepoint = savepointStore.getState(savepointPath);
- verifySavepoint(savepoint, jobId, checkpointId, timestamp,
- vertices);
-
- // Verify all promises removed
- assertEquals(0, getSavepointPromises(coordinator).size());
-
- coordinator.shutdown();
- }
-
- /**
- * This test triggers a checkpoint and then sends a decline checkpoint message from
- * one of the tasks. The expected behaviour is that said checkpoint is discarded and a new
- * checkpoint is triggered.
- */
- @Test
- public void testTriggerAndDeclineCheckpointSimple() throws Exception {
- JobID jobId = new JobID();
- long checkpointTimeout = 60 * 1000;
- long timestamp = 1272635;
- ExecutionVertex[] vertices = new ExecutionVertex[] {
- mockExecutionVertex(jobId),
- mockExecutionVertex(jobId) };
- MockCheckpointIdCounter checkpointIdCounter = new MockCheckpointIdCounter();
- HeapStateStore<CompletedCheckpoint> savepointStore = new HeapStateStore<>();
-
- SavepointCoordinator coordinator = createSavepointCoordinator(
- jobId,
- checkpointTimeout,
- vertices,
- vertices,
- vertices,
- checkpointIdCounter,
- savepointStore);
-
- // Trigger the savepoint
- Future<String> savepointPathFuture = coordinator.triggerSavepoint(timestamp);
- assertFalse(savepointPathFuture.isCompleted());
-
- long checkpointId = checkpointIdCounter.getLastReturnedCount();
- assertEquals(0, checkpointId);
-
- // Verify send trigger messages
- for (ExecutionVertex vertex : vertices) {
- verifyTriggerCheckpoint(vertex, checkpointId, timestamp);
- }
-
- PendingCheckpoint pendingCheckpoint = coordinator.getPendingCheckpoints()
- .get(checkpointId);
-
- verifyPendingCheckpoint(pendingCheckpoint, jobId, checkpointId,
- timestamp, 0, 2, 0, false, false);
-
- // Acknowledge and decline tasks
- coordinator.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(
- jobId, vertices[0].getCurrentExecutionAttempt().getAttemptId(),
- checkpointId, createSerializedStateHandle(vertices[0]), 0));
-
- coordinator.receiveDeclineMessage(new DeclineCheckpoint(
- jobId, vertices[1].getCurrentExecutionAttempt().getAttemptId(),
- checkpointId, 0));
-
-
- // The pending checkpoint is completed
- assertTrue(pendingCheckpoint.isDiscarded());
- assertEquals(0, coordinator.getSuccessfulCheckpoints().size());
-
- // Verify that the future has been completed
- assertTrue(savepointPathFuture.isCompleted());
-
- try {
- Await.result(savepointPathFuture.failed(), FiniteDuration.Zero());
- fail("Did not throw expected exception");
- } catch (Throwable ignored) {}
-
- // Verify all promises removed
- assertEquals(0, getSavepointPromises(coordinator).size());
-
- coordinator.shutdown();
- }
-
- // ------------------------------------------------------------------------
- // Rollback
- // ------------------------------------------------------------------------
-
- @Test
- @SuppressWarnings("unchecked")
- public void testSimpleRollbackSavepoint() throws Exception {
- JobID jobId = new JobID();
-
- ExecutionJobVertex[] jobVertices = new ExecutionJobVertex[] {
- mockExecutionJobVertex(jobId, new JobVertexID(), 4),
- mockExecutionJobVertex(jobId, new JobVertexID(), 4) };
-
- ExecutionVertex[] triggerVertices = jobVertices[0].getTaskVertices();
- ExecutionVertex[] ackVertices = new ExecutionVertex[8];
-
- int i = 0;
- for (ExecutionJobVertex jobVertex : jobVertices) {
- for (ExecutionVertex vertex : jobVertex.getTaskVertices()) {
- ackVertices[i++] = vertex;
- }
- }
-
- MockCheckpointIdCounter idCounter = new MockCheckpointIdCounter();
- StateStore<CompletedCheckpoint> savepointStore = new HeapStateStore<>();
-
- SavepointCoordinator coordinator = createSavepointCoordinator(
- jobId,
- 60 * 1000,
- triggerVertices,
- ackVertices,
- new ExecutionVertex[] {},
- idCounter,
- savepointStore);
-
- Future<String> savepointPathFuture = coordinator.triggerSavepoint(1231273123);
-
- // Acknowledge all tasks
- for (ExecutionVertex vertex : ackVertices) {
- ExecutionAttemptID attemptId = vertex.getCurrentExecutionAttempt().getAttemptId();
- coordinator.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(
- jobId, attemptId, 0, createSerializedStateHandle(vertex), 0));
- }
-
- String savepointPath = Await.result(savepointPathFuture, FiniteDuration.Zero());
- assertNotNull(savepointPath);
-
- // Rollback
- coordinator.restoreSavepoint(createExecutionJobVertexMap(jobVertices), savepointPath);
-
- // Verify all executions have been reset
- for (ExecutionVertex vertex : ackVertices) {
- verify(vertex.getCurrentExecutionAttempt(), times(1)).setInitialState(
- any(SerializedValue.class), any(Map.class));
- }
-
- // Verify all promises removed
- assertEquals(0, getSavepointPromises(coordinator).size());
-
- // Verify checkpoint ID counter started
- assertTrue(idCounter.isStarted());
-
- coordinator.shutdown();
- }
-
- @Test
- public void testRollbackParallelismMismatch() throws Exception {
- JobID jobId = new JobID();
-
- ExecutionJobVertex[] jobVertices = new ExecutionJobVertex[] {
- mockExecutionJobVertex(jobId, new JobVertexID(), 4),
- mockExecutionJobVertex(jobId, new JobVertexID(), 4) };
-
- ExecutionVertex[] triggerVertices = jobVertices[0].getTaskVertices();
- ExecutionVertex[] ackVertices = new ExecutionVertex[8];
-
- int index = 0;
- for (ExecutionJobVertex jobVertex : jobVertices) {
- for (ExecutionVertex vertex : jobVertex.getTaskVertices()) {
- ackVertices[index++] = vertex;
- }
- }
-
- StateStore<CompletedCheckpoint> savepointStore = new HeapStateStore<>();
-
- SavepointCoordinator coordinator = createSavepointCoordinator(
- jobId,
- 60 * 1000,
- triggerVertices,
- ackVertices,
- new ExecutionVertex[] {},
- new MockCheckpointIdCounter(),
- savepointStore);
-
- Future<String> savepointPathFuture = coordinator.triggerSavepoint(1231273123);
-
- // Acknowledge all tasks
- for (ExecutionVertex vertex : ackVertices) {
- ExecutionAttemptID attemptId = vertex.getCurrentExecutionAttempt().getAttemptId();
- coordinator.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(
- jobId, attemptId, 0, createSerializedStateHandle(vertex), 0));
- }
-
- String savepointPath = Await.result(savepointPathFuture, FiniteDuration.Zero());
- assertNotNull(savepointPath);
-
- // 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);
- }
-
- try {
- // Rollback
- coordinator.restoreSavepoint(
- createExecutionJobVertexMap(jobVertices),
- savepointPath);
- fail("Did not throw expected Exception after rollback with parallelism mismatch.");
- }
- catch (Exception ignored) {
- }
-
- // Verify all promises removed
- assertEquals(0, getSavepointPromises(coordinator).size());
-
- coordinator.shutdown();
- }
-
- @Test
- public void testRollbackStateStoreFailure() throws Exception {
- JobID jobId = new JobID();
- ExecutionJobVertex jobVertex = mockExecutionJobVertex(jobId, new JobVertexID(), 4);
- HeapStateStore<CompletedCheckpoint> savepointStore = spy(
- new HeapStateStore<CompletedCheckpoint>());
-
- SavepointCoordinator coordinator = createSavepointCoordinator(
- jobId,
- 60 * 1000,
- jobVertex.getTaskVertices(),
- jobVertex.getTaskVertices(),
- new ExecutionVertex[] {},
- new MockCheckpointIdCounter(),
- savepointStore);
-
- Future<String> savepointPathFuture = coordinator.triggerSavepoint(1231273123);
-
- // Acknowledge all tasks
- for (ExecutionVertex vertex : jobVertex.getTaskVertices()) {
- ExecutionAttemptID attemptId = vertex.getCurrentExecutionAttempt().getAttemptId();
- coordinator.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(
- jobId, attemptId, 0, createSerializedStateHandle(vertex), 0));
- }
-
- String savepointPath = Await.result(savepointPathFuture, FiniteDuration.Zero());
- assertNotNull(savepointPath);
-
- // Failure on getState
- doThrow(new Exception("TestException")).when(savepointStore).getState(anyString());
-
- try {
- // Rollback
- coordinator.restoreSavepoint(
- createExecutionJobVertexMap(jobVertex),
- savepointPath);
-
- fail("Did not throw expected Exception after rollback with savepoint store failure.");
- }
- catch (Exception ignored) {
- }
-
- // Verify all promises removed
- assertEquals(0, getSavepointPromises(coordinator).size());
-
- coordinator.shutdown();
- }
-
- @Test
- public void testRollbackSetsCheckpointID() throws Exception {
- CompletedCheckpoint savepoint = mock(CompletedCheckpoint.class);
- when(savepoint.getTaskStates()).thenReturn(Collections.<JobVertexID, TaskState>emptyMap());
- when(savepoint.getCheckpointID()).thenReturn(12312312L);
-
- CheckpointIDCounter checkpointIdCounter = mock(CheckpointIDCounter.class);
-
- StateStore<CompletedCheckpoint> savepointStore = mock(StateStore.class);
- when(savepointStore.getState(anyString())).thenReturn(savepoint);
-
- SavepointCoordinator coordinator = createSavepointCoordinator(
- new JobID(),
- 60 * 1000,
- new ExecutionVertex[] {},
- new ExecutionVertex[] {},
- new ExecutionVertex[] {},
- checkpointIdCounter,
- savepointStore);
-
- coordinator.restoreSavepoint(createExecutionJobVertexMap(), "any");
-
- verify(checkpointIdCounter).setCount(eq(12312312L + 1));
-
- coordinator.shutdown();
- }
-
- // ------------------------------------------------------------------------
- // Savepoint aborts and future notifications
- // ------------------------------------------------------------------------
-
- @Test
- public void testAbortSavepointIfTriggerTasksNotExecuted() throws Exception {
- JobID jobId = new JobID();
- ExecutionVertex[] triggerVertices = new ExecutionVertex[] {
- mock(ExecutionVertex.class),
- mock(ExecutionVertex.class) };
- ExecutionVertex[] ackVertices = new ExecutionVertex[] {
- mockExecutionVertex(jobId),
- mockExecutionVertex(jobId) };
-
- SavepointCoordinator coordinator = createSavepointCoordinator(
- jobId,
- 60 * 1000,
- triggerVertices,
- ackVertices,
- new ExecutionVertex[] {},
- new MockCheckpointIdCounter(),
- new HeapStateStore<CompletedCheckpoint>());
-
- // Trigger savepoint
- Future<String> savepointPathFuture = coordinator.triggerSavepoint(1238123);
-
- // Abort the savepoint, because the vertices are not running
- assertTrue(savepointPathFuture.isCompleted());
-
- try {
- Await.result(savepointPathFuture, FiniteDuration.Zero());
- fail("Did not throw expected Exception after shutdown");
- }
- catch (Exception ignored) {
- }
-
- // Verify all promises removed
- assertEquals(0, getSavepointPromises(coordinator).size());
-
- coordinator.shutdown();
- }
-
- @Test
- public void testAbortSavepointIfTriggerTasksAreFinished() throws Exception {
- JobID jobId = new JobID();
- ExecutionVertex[] triggerVertices = new ExecutionVertex[] {
- mockExecutionVertex(jobId),
- mockExecutionVertex(jobId, ExecutionState.FINISHED) };
- ExecutionVertex[] ackVertices = new ExecutionVertex[] {
- mockExecutionVertex(jobId),
- mockExecutionVertex(jobId) };
-
- SavepointCoordinator coordinator = createSavepointCoordinator(
- jobId,
- 60 * 1000,
- triggerVertices,
- ackVertices,
- new ExecutionVertex[] {},
- new MockCheckpointIdCounter(),
- new HeapStateStore<CompletedCheckpoint>());
-
- // Trigger savepoint
- Future<String> savepointPathFuture = coordinator.triggerSavepoint(1238123);
-
- // Abort the savepoint, because the vertices are not running
- assertTrue(savepointPathFuture.isCompleted());
-
- try {
- Await.result(savepointPathFuture, FiniteDuration.Zero());
- fail("Did not throw expected Exception after shutdown");
- }
- catch (Exception ignored) {
- }
-
- // Verify all promises removed
- assertEquals(0, getSavepointPromises(coordinator).size());
-
- coordinator.shutdown();
- }
-
- @Test
- public void testAbortSavepointIfAckTasksAreNotExecuted() throws Exception {
- JobID jobId = new JobID();
- ExecutionVertex[] triggerVertices = new ExecutionVertex[] {
- mockExecutionVertex(jobId),
- mockExecutionVertex(jobId) };
- ExecutionVertex[] ackVertices = new ExecutionVertex[] {
- mock(ExecutionVertex.class),
- mock(ExecutionVertex.class) };
-
- SavepointCoordinator coordinator = createSavepointCoordinator(
- jobId,
- 60 * 1000,
- triggerVertices,
- ackVertices,
- new ExecutionVertex[] {},
- new MockCheckpointIdCounter(),
- new HeapStateStore<CompletedCheckpoint>());
-
- // Trigger savepoint
- Future<String> savepointPathFuture = coordinator.triggerSavepoint(1238123);
-
- // Abort the savepoint, because the vertices are not running
- assertTrue(savepointPathFuture.isCompleted());
-
- try {
- Await.result(savepointPathFuture, FiniteDuration.Zero());
- fail("Did not throw expected Exception after shutdown");
- }
- catch (Exception ignored) {
- }
-
- // Verify all promises removed
- assertEquals(0, getSavepointPromises(coordinator).size());
-
- coordinator.shutdown();
- }
-
- @Test
- public void testAbortOnCheckpointTimeout() throws Exception {
- JobID jobId = new JobID();
- ExecutionVertex[] vertices = new ExecutionVertex[] {
- mockExecutionVertex(jobId),
- mockExecutionVertex(jobId) };
- ExecutionVertex commitVertex = mockExecutionVertex(jobId);
- MockCheckpointIdCounter checkpointIdCounter = new MockCheckpointIdCounter();
-
- long checkpointTimeout = 1000;
- SavepointCoordinator coordinator = createSavepointCoordinator(
- jobId,
- checkpointTimeout,
- vertices,
- vertices,
- new ExecutionVertex[] { commitVertex },
- checkpointIdCounter,
- new HeapStateStore<CompletedCheckpoint>());
-
- // Trigger the savepoint
- Future<String> savepointPathFuture = coordinator.triggerSavepoint(12731273);
- assertFalse(savepointPathFuture.isCompleted());
-
- long checkpointId = checkpointIdCounter.getLastReturnedCount();
- PendingCheckpoint pendingCheckpoint = coordinator.getPendingCheckpoints()
- .get(checkpointId);
-
- assertNotNull("Checkpoint not pending (test race)", pendingCheckpoint);
- assertFalse("Checkpoint already discarded (test race)", pendingCheckpoint.isDiscarded());
-
- // Wait for savepoint to timeout
- Deadline deadline = FiniteDuration.apply(60, "s").fromNow();
- while (deadline.hasTimeLeft()
- && !pendingCheckpoint.isDiscarded()
- && coordinator.getNumberOfPendingCheckpoints() > 0) {
-
- Thread.sleep(250);
- }
-
- // Verify discarded
- assertTrue("Savepoint not discarded within timeout", pendingCheckpoint.isDiscarded());
- assertEquals(0, coordinator.getNumberOfPendingCheckpoints());
- assertEquals(0, coordinator.getNumberOfRetainedSuccessfulCheckpoints());
-
- // No commit for timeout
- verify(commitVertex, times(0)).sendMessageToCurrentExecution(
- any(NotifyCheckpointComplete.class), any(ExecutionAttemptID.class));
-
- assertTrue(savepointPathFuture.isCompleted());
-
- try {
- Await.result(savepointPathFuture, FiniteDuration.Zero());
- fail("Did not throw expected Exception after timeout");
- }
- catch (Exception ignored) {
- }
-
- // Verify all promises removed
- assertEquals(0, getSavepointPromises(coordinator).size());
-
- coordinator.shutdown();
- }
-
- @Test
- public void testAbortSavepointsOnShutdown() throws Exception {
- JobID jobId = new JobID();
- ExecutionVertex[] vertices = new ExecutionVertex[] {
- mockExecutionVertex(jobId),
- mockExecutionVertex(jobId) };
-
- SavepointCoordinator coordinator = createSavepointCoordinator(
- jobId,
- 60 * 1000,
- vertices,
- vertices,
- vertices,
- new MockCheckpointIdCounter(),
- new HeapStateStore<CompletedCheckpoint>());
-
- // Trigger savepoints
- List<Future<String>> savepointPathFutures = new ArrayList<>();
- savepointPathFutures.add(coordinator.triggerSavepoint(12731273));
- savepointPathFutures.add(coordinator.triggerSavepoint(12731273 + 123));
-
- for (Future<String> future : savepointPathFutures) {
- assertFalse(future.isCompleted());
- }
-
- coordinator.shutdown();
-
- // Verify futures failed
- for (Future<String> future : savepointPathFutures) {
- assertTrue(future.isCompleted());
-
- try {
- Await.result(future, FiniteDuration.Zero());
- fail("Did not throw expected Exception after shutdown");
- }
- catch (Exception ignored) {
- }
- }
-
- // Verify all promises removed
- assertEquals(0, getSavepointPromises(coordinator).size());
- }
-
- @Test
- public void testAbortSavepointOnStateStoreFailure() throws Exception {
- JobID jobId = new JobID();
- ExecutionJobVertex jobVertex = mockExecutionJobVertex(jobId, new JobVertexID(), 4);
- HeapStateStore<CompletedCheckpoint> savepointStore = spy(
- new HeapStateStore<CompletedCheckpoint>());
-
- SavepointCoordinator coordinator = createSavepointCoordinator(
- jobId,
- 60 * 1000,
- jobVertex.getTaskVertices(),
- jobVertex.getTaskVertices(),
- new ExecutionVertex[] {},
- new MockCheckpointIdCounter(),
- savepointStore);
-
- // Failure on putState
- doThrow(new Exception("TestException"))
- .when(savepointStore).putState(any(CompletedCheckpoint.class));
-
- Future<String> savepointPathFuture = coordinator.triggerSavepoint(1231273123);
-
- // Acknowledge all tasks
- for (ExecutionVertex vertex : jobVertex.getTaskVertices()) {
- ExecutionAttemptID attemptId = vertex.getCurrentExecutionAttempt().getAttemptId();
- coordinator.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(
- jobId, attemptId, 0, createSerializedStateHandle(vertex), 0));
- }
-
- try {
- Await.result(savepointPathFuture, FiniteDuration.Zero());
- fail("Did not throw expected Exception after rollback with savepoint store failure.");
- }
- catch (Exception ignored) {
- }
-
- // Verify all promises removed
- assertEquals(0, getSavepointPromises(coordinator).size());
-
- coordinator.shutdown();
- }
-
- @Test
- public void testAbortSavepointIfSubsumed() throws Exception {
- JobID jobId = new JobID();
- long checkpointTimeout = 60 * 1000;
- long[] timestamps = new long[] { 1272635, 1272635 + 10 };
- long[] checkpointIds = new long[2];
- ExecutionVertex[] vertices = new ExecutionVertex[] {
- mockExecutionVertex(jobId),
- mockExecutionVertex(jobId) };
- MockCheckpointIdCounter checkpointIdCounter = new MockCheckpointIdCounter();
- HeapStateStore<CompletedCheckpoint> savepointStore = new HeapStateStore<>();
-
- SavepointCoordinator coordinator = createSavepointCoordinator(
- jobId,
- checkpointTimeout,
- vertices,
- vertices,
- vertices,
- checkpointIdCounter,
- savepointStore);
-
- // Trigger the savepoints
- List<Future<String>> savepointPathFutures = new ArrayList<>();
-
- savepointPathFutures.add(coordinator.triggerSavepoint(timestamps[0]));
- checkpointIds[0] = checkpointIdCounter.getLastReturnedCount();
-
- savepointPathFutures.add(coordinator.triggerSavepoint(timestamps[1]));
- checkpointIds[1] = checkpointIdCounter.getLastReturnedCount();
-
- for (Future<String> future : savepointPathFutures) {
- assertFalse(future.isCompleted());
- }
-
- // Verify send trigger messages
- for (ExecutionVertex vertex : vertices) {
- verifyTriggerCheckpoint(vertex, checkpointIds[0], timestamps[0]);
- verifyTriggerCheckpoint(vertex, checkpointIds[1], timestamps[1]);
- }
-
- PendingCheckpoint[] pendingCheckpoints = new PendingCheckpoint[] {
- coordinator.getPendingCheckpoints().get(checkpointIds[0]),
- coordinator.getPendingCheckpoints().get(checkpointIds[1]) };
-
- verifyPendingCheckpoint(pendingCheckpoints[0], jobId, checkpointIds[0],
- timestamps[0], 0, 2, 0, false, false);
-
- verifyPendingCheckpoint(pendingCheckpoints[1], jobId, checkpointIds[1],
- timestamps[1], 0, 2, 0, false, false);
-
- // Acknowledge second checkpoint...
- for (ExecutionVertex vertex : vertices) {
- coordinator.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(
- jobId, vertex.getCurrentExecutionAttempt().getAttemptId(),
- checkpointIds[1], createSerializedStateHandle(vertex), 0));
- }
-
- // ...and one task of first checkpoint
- coordinator.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(
- jobId, vertices[0].getCurrentExecutionAttempt().getAttemptId(),
- checkpointIds[0], createSerializedStateHandle(vertices[0]), 0));
-
- // The second pending checkpoint is completed and subsumes the first one
- assertTrue(pendingCheckpoints[0].isDiscarded());
- assertTrue(pendingCheckpoints[1].isDiscarded());
- assertEquals(0, coordinator.getSuccessfulCheckpoints().size());
-
- // Verify send notify complete messages for second checkpoint
- for (ExecutionVertex vertex : vertices) {
- verifyNotifyCheckpointComplete(vertex, checkpointIds[1], timestamps[1]);
- }
-
- CompletedCheckpoint[] savepoints = new CompletedCheckpoint[2];
- String[] savepointPaths = new String[2];
-
- // Verify that the futures have both been completed
- assertTrue(savepointPathFutures.get(0).isCompleted());
-
- try {
- savepointPaths[0] = Await.result(savepointPathFutures.get(0), FiniteDuration.Zero());
- fail("Did not throw expected exception");
- }
- catch (Exception ignored) {
- }
-
- // Verify the second savepoint
- assertTrue(savepointPathFutures.get(1).isCompleted());
- savepointPaths[1] = Await.result(savepointPathFutures.get(1), FiniteDuration.Zero());
- savepoints[1] = savepointStore.getState(savepointPaths[1]);
- verifySavepoint(savepoints[1], jobId, checkpointIds[1], timestamps[1],
- vertices);
-
- // Verify all promises removed
- assertEquals(0, getSavepointPromises(coordinator).size());
-
- coordinator.shutdown();
- }
-
- @Test
- public void testShutdownDoesNotCleanUpCompletedCheckpointsWithFileSystemStore() throws Exception {
- JobID jobId = new JobID();
- long checkpointTimeout = 60 * 1000;
- long timestamp = 1272635;
- ExecutionVertex[] vertices = new ExecutionVertex[] {
- mockExecutionVertex(jobId),
- mockExecutionVertex(jobId) };
- MockCheckpointIdCounter checkpointIdCounter = new MockCheckpointIdCounter();
-
- // Temporary directory for file state backend
- final File tmpDir = CommonTestUtils.createTempDirectory();
-
- try {
- FileSystemStateStore<CompletedCheckpoint> savepointStore = new FileSystemStateStore<>(
- tmpDir.toURI().toString(), "sp-");
-
- SavepointCoordinator coordinator = createSavepointCoordinator(
- jobId,
- checkpointTimeout,
- vertices,
- vertices,
- vertices,
- checkpointIdCounter,
- savepointStore);
-
- // Trigger the savepoint
- Future<String> savepointPathFuture = coordinator.triggerSavepoint(timestamp);
- assertFalse(savepointPathFuture.isCompleted());
-
- long checkpointId = checkpointIdCounter.getLastReturnedCount();
- assertEquals(0, checkpointId);
-
- // Verify send trigger messages
- for (ExecutionVertex vertex : vertices) {
- verifyTriggerCheckpoint(vertex, checkpointId, timestamp);
- }
-
- PendingCheckpoint pendingCheckpoint = coordinator.getPendingCheckpoints()
- .get(checkpointId);
-
- verifyPendingCheckpoint(pendingCheckpoint, jobId, checkpointId,
- timestamp, 0, 2, 0, false, false);
-
- // Acknowledge tasks
- for (ExecutionVertex vertex : vertices) {
- coordinator.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(
- jobId, vertex.getCurrentExecutionAttempt().getAttemptId(),
- checkpointId, createSerializedStateHandle(vertex), 0));
- }
-
- // The pending checkpoint is completed
- assertTrue(pendingCheckpoint.isDiscarded());
- assertEquals(0, coordinator.getSuccessfulCheckpoints().size());
-
- // Verify send notify complete messages
- for (ExecutionVertex vertex : vertices) {
- verifyNotifyCheckpointComplete(vertex, checkpointId, timestamp);
- }
-
- // Verify that the future has been completed
- assertTrue(savepointPathFuture.isCompleted());
- String savepointPath = Await.result(savepointPathFuture, FiniteDuration.Zero());
-
- // Verify all promises removed
- assertEquals(0, getSavepointPromises(coordinator).size());
-
- coordinator.shutdown();
-
- // Verify the savepoint is still available
- CompletedCheckpoint savepoint = savepointStore.getState(savepointPath);
- verifySavepoint(savepoint, jobId, checkpointId, timestamp,
- vertices);
- }
- finally {
- FileUtils.deleteDirectory(tmpDir);
- }
- }
-
- // ------------------------------------------------------------------------
- // Test helpers
- // ------------------------------------------------------------------------
-
- private static SavepointCoordinator createSavepointCoordinator(
- JobID jobId,
- long checkpointTimeout,
- ExecutionVertex[] triggerVertices,
- ExecutionVertex[] ackVertices,
- ExecutionVertex[] commitVertices,
- CheckpointIDCounter checkpointIdCounter,
- StateStore<CompletedCheckpoint> savepointStore) throws Exception {
-
- ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
-
- return new SavepointCoordinator(
- jobId,
- checkpointTimeout,
- checkpointTimeout,
- 42,
- triggerVertices,
- ackVertices,
- commitVertices,
- classLoader,
- checkpointIdCounter,
- savepointStore,
- new DisabledCheckpointStatsTracker());
- }
-
- private static Map<JobVertexID, ExecutionJobVertex> createExecutionJobVertexMap(
- ExecutionJobVertex... jobVertices) {
-
- Map<JobVertexID, ExecutionJobVertex> jobVertexMap = new HashMap<>();
-
- for (ExecutionJobVertex jobVertex : jobVertices) {
- jobVertexMap.put(jobVertex.getJobVertexId(), jobVertex);
- }
-
- return jobVertexMap;
- }
-
- private static SerializedValue<StateHandle<?>> createSerializedStateHandle(
- ExecutionVertex vertex) throws IOException {
-
- return new SerializedValue<StateHandle<?>>(new LocalStateHandle<Serializable>(
- vertex.getCurrentExecutionAttempt().getAttemptId()));
- }
-
- @SuppressWarnings("unchecked")
- private Map<Long, Promise<String>> getSavepointPromises(
- SavepointCoordinator coordinator)
- throws NoSuchFieldException, IllegalAccessException {
-
- Field field = SavepointCoordinator.class.getDeclaredField("savepointPromises");
- field.setAccessible(true);
- return (Map<Long, Promise<String>>) field.get(coordinator);
- }
-
- // ---- Verification ------------------------------------------------------
-
- private static void verifyTriggerCheckpoint(
- ExecutionVertex mockExecutionVertex,
- long expectedCheckpointId,
- long expectedTimestamp) {
-
- ExecutionAttemptID attemptId = mockExecutionVertex
- .getCurrentExecutionAttempt().getAttemptId();
-
- TriggerCheckpoint expectedMsg = new TriggerCheckpoint(
- mockExecutionVertex.getJobId(),
- attemptId,
- expectedCheckpointId,
- expectedTimestamp);
-
- verify(mockExecutionVertex).sendMessageToCurrentExecution(
- eq(expectedMsg), eq(attemptId));
- }
-
- private static void verifyNotifyCheckpointComplete(
- ExecutionVertex mockExecutionVertex,
- long expectedCheckpointId,
- long expectedTimestamp) {
-
- ExecutionAttemptID attemptId = mockExecutionVertex
- .getCurrentExecutionAttempt().getAttemptId();
-
- NotifyCheckpointComplete expectedMsg = new NotifyCheckpointComplete(
- mockExecutionVertex.getJobId(),
- attemptId,
- expectedCheckpointId,
- expectedTimestamp);
-
- verify(mockExecutionVertex).sendMessageToCurrentExecution(
- eq(expectedMsg), eq(attemptId));
- }
-
- private static void verifyPendingCheckpoint(
- PendingCheckpoint checkpoint,
- JobID expectedJobId,
- long expectedCheckpointId,
- long expectedTimestamp,
- int expectedNumberOfAcknowledgedTasks,
- int expectedNumberOfNonAcknowledgedTasks,
- int expectedNumberOfCollectedStates,
- boolean expectedIsDiscarded,
- boolean expectedIsFullyAcknowledged) {
-
- assertNotNull(checkpoint);
- assertEquals(expectedJobId, checkpoint.getJobId());
- assertEquals(expectedCheckpointId, checkpoint.getCheckpointId());
- assertEquals(expectedTimestamp, checkpoint.getCheckpointTimestamp());
- assertEquals(expectedNumberOfAcknowledgedTasks, checkpoint.getNumberOfAcknowledgedTasks());
- assertEquals(expectedNumberOfNonAcknowledgedTasks, checkpoint.getNumberOfNonAcknowledgedTasks());
-
- int actualNumberOfCollectedStates = 0;
-
- for (TaskState taskState : checkpoint.getTaskStates().values()) {
- actualNumberOfCollectedStates += taskState.getNumberCollectedStates();
- }
-
- assertEquals(expectedNumberOfCollectedStates, actualNumberOfCollectedStates);
- assertEquals(expectedIsDiscarded, checkpoint.isDiscarded());
- assertEquals(expectedIsFullyAcknowledged, checkpoint.isFullyAcknowledged());
- }
-
- private static void verifySavepoint(
- CompletedCheckpoint savepoint,
- JobID expectedJobId,
- long expectedCheckpointId,
- long expectedTimestamp,
- ExecutionVertex[] expectedVertices) throws Exception {
-
- verifyCompletedCheckpoint(
- savepoint,
- expectedJobId,
- expectedCheckpointId,
- expectedTimestamp,
- expectedVertices
- );
- }
-
- private static void verifyCompletedCheckpoint(
- CompletedCheckpoint checkpoint,
- JobID expectedJobId,
- long expectedCheckpointId,
- long expectedTimestamp,
- ExecutionVertex[] expectedVertices) throws Exception {
-
- assertNotNull(checkpoint);
- assertEquals(expectedJobId, checkpoint.getJobId());
- assertEquals(expectedCheckpointId, checkpoint.getCheckpointID());
- assertEquals(expectedTimestamp, checkpoint.getTimestamp());
-
- for (ExecutionVertex vertex : expectedVertices) {
- JobVertexID jobVertexID = vertex.getJobvertexId();
-
- TaskState taskState = checkpoint.getTaskState(jobVertexID);
-
- assertNotNull(taskState);
-
- SubtaskState subtaskState = taskState.getState(vertex.getParallelSubtaskIndex());
-
- assertNotNull(subtaskState);
-
- ExecutionAttemptID vertexAttemptId = vertex.getCurrentExecutionAttempt().getAttemptId();
-
- ExecutionAttemptID stateAttemptId = (ExecutionAttemptID) subtaskState.getState()
- .deserializeValue(Thread.currentThread().getContextClassLoader())
- .getState(Thread.currentThread().getContextClassLoader());
-
- assertEquals(vertexAttemptId, stateAttemptId);
- }
- }
-
- // ---- Mocking -----------------------------------------------------------
-
- private static ExecutionJobVertex mockExecutionJobVertex(
- JobID jobId,
- JobVertexID jobVertexId,
- int parallelism) {
-
- ExecutionJobVertex jobVertex = mock(ExecutionJobVertex.class);
- when(jobVertex.getJobId()).thenReturn(jobId);
- when(jobVertex.getJobVertexId()).thenReturn(jobVertexId);
- when(jobVertex.getParallelism()).thenReturn(parallelism);
-
- ExecutionVertex[] vertices = new ExecutionVertex[parallelism];
-
- for (int i = 0; i < vertices.length; i++) {
- vertices[i] = mockExecutionVertex(jobId, jobVertexId, i, parallelism, ExecutionState.RUNNING);
- }
-
- when(jobVertex.getTaskVertices()).thenReturn(vertices);
-
- return jobVertex;
- }
-
- private static ExecutionVertex mockExecutionVertex(JobID jobId) {
- return mockExecutionVertex(jobId, ExecutionState.RUNNING);
- }
-
- private static ExecutionVertex mockExecutionVertex(
- JobID jobId,
- ExecutionState state) {
-
- return mockExecutionVertex(jobId, new JobVertexID(), 0, 1, state);
- }
-
- private static ExecutionVertex mockExecutionVertex(
- JobID jobId,
- JobVertexID jobVertexId,
- int subtaskIndex,
- int parallelism,
- ExecutionState executionState) {
-
- Execution exec = mock(Execution.class);
- when(exec.getAttemptId()).thenReturn(new ExecutionAttemptID());
- when(exec.getState()).thenReturn(executionState);
-
- ExecutionVertex vertex = mock(ExecutionVertex.class);
- when(vertex.getJobId()).thenReturn(jobId);
- when(vertex.getJobvertexId()).thenReturn(jobVertexId);
- when(vertex.getParallelSubtaskIndex()).thenReturn(subtaskIndex);
- when(vertex.getCurrentExecutionAttempt()).thenReturn(exec);
- when(vertex.getTotalNumberOfParallelSubtasks()).thenReturn(parallelism);
-
- return vertex;
- }
-
- private static class MockCheckpointIdCounter implements CheckpointIDCounter {
-
- private boolean started;
- private long count;
- private long lastReturnedCount;
-
- @Override
- public void start() throws Exception {
- started = true;
- }
-
- @Override
- public void shutdown() throws Exception {
- started = false;
- }
-
- @Override
- public void suspend() throws Exception {
- started = false;
- }
-
- @Override
- public long getAndIncrement() throws Exception {
- lastReturnedCount = count;
- return count++;
- }
-
- @Override
- public void setCount(long newCount) {
- count = newCount;
- }
-
- long getLastReturnedCount() {
- return lastReturnedCount;
- }
-
- public boolean isStarted() {
- return started;
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/SavepointStoreFactoryTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/SavepointStoreFactoryTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/SavepointStoreFactoryTest.java
deleted file mode 100644
index c0605f7..0000000
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/SavepointStoreFactoryTest.java
+++ /dev/null
@@ -1,96 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.checkpoint;
-
-import org.apache.flink.configuration.ConfigConstants;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.IllegalConfigurationException;
-import org.apache.flink.core.fs.Path;
-import org.junit.Test;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
-
-public class SavepointStoreFactoryTest {
-
- @Test
- public void testStateStoreWithDefaultConfig() throws Exception {
- SavepointStore store = SavepointStoreFactory.createFromConfig(new Configuration());
- assertTrue(store.getStateStore() instanceof HeapStateStore);
- }
-
- @Test
- public void testSavepointBackendJobManager() throws Exception {
- Configuration config = new Configuration();
- config.setString(SavepointStoreFactory.SAVEPOINT_BACKEND_KEY, "jobmanager");
- SavepointStore store = SavepointStoreFactory.createFromConfig(config);
- assertTrue(store.getStateStore() instanceof HeapStateStore);
- }
-
- @Test
- public void testSavepointBackendFileSystem() throws Exception {
- Configuration config = new Configuration();
- String rootPath = System.getProperty("java.io.tmpdir");
- config.setString(ConfigConstants.STATE_BACKEND, "filesystem");
- config.setString(SavepointStoreFactory.SAVEPOINT_BACKEND_KEY, "filesystem");
- config.setString(SavepointStoreFactory.SAVEPOINT_DIRECTORY_KEY, rootPath);
-
- SavepointStore store = SavepointStoreFactory.createFromConfig(config);
- assertTrue(store.getStateStore() instanceof FileSystemStateStore);
-
- FileSystemStateStore<CompletedCheckpoint> stateStore = (FileSystemStateStore<CompletedCheckpoint>)
- store.getStateStore();
- assertEquals(new Path(rootPath), stateStore.getRootPath());
- }
-
- @Test
- public void testSavepointBackendFileSystemButCheckpointBackendJobManager() throws Exception {
- Configuration config = new Configuration();
- String rootPath = System.getProperty("java.io.tmpdir");
- // This combination does not make sense, because the checkpoints will be
- // lost after the job manager shuts down.
- config.setString(ConfigConstants.STATE_BACKEND, "jobmanager");
- config.setString(SavepointStoreFactory.SAVEPOINT_BACKEND_KEY, "filesystem");
- config.setString(SavepointStoreFactory.SAVEPOINT_DIRECTORY_KEY, rootPath);
-
- SavepointStore store = SavepointStoreFactory.createFromConfig(config);
- assertTrue(store.getStateStore() instanceof FileSystemStateStore);
-
- FileSystemStateStore<CompletedCheckpoint> stateStore = (FileSystemStateStore<CompletedCheckpoint>)
- store.getStateStore();
- assertEquals(new Path(rootPath), stateStore.getRootPath());
- }
-
- @Test(expected = IllegalConfigurationException.class)
- public void testSavepointBackendFileSystemButNoDirectory() throws Exception {
- Configuration config = new Configuration();
- config.setString(SavepointStoreFactory.SAVEPOINT_BACKEND_KEY, "filesystem");
- SavepointStoreFactory.createFromConfig(config);
- fail("Did not throw expected Exception");
- }
-
- @Test(expected = IllegalConfigurationException.class)
- public void testUnexpectedSavepointBackend() throws Exception {
- Configuration config = new Configuration();
- config.setString(SavepointStoreFactory.SAVEPOINT_BACKEND_KEY, "unexpected");
- SavepointStoreFactory.createFromConfig(config);
- fail("Did not throw expected Exception");
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0ac261a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/StandaloneCompletedCheckpointStoreTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/StandaloneCompletedCheckpointStoreTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/StandaloneCompletedCheckpointStoreTest.java
index 6e4ffd6..59fa0e2 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/StandaloneCompletedCheckpointStoreTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/StandaloneCompletedCheckpointStoreTest.java
@@ -42,7 +42,7 @@ public class StandaloneCompletedCheckpointStoreTest extends CompletedCheckpointS
@Test
public void testShutdownDiscardsCheckpoints() throws Exception {
CompletedCheckpointStore store = createCompletedCheckpoints(1, ClassLoader.getSystemClassLoader());
- TestCheckpoint checkpoint = createCheckpoint(0);
+ TestCompletedCheckpoint checkpoint = createCheckpoint(0);
store.addCheckpoint(checkpoint);
assertEquals(1, store.getNumberOfRetainedCheckpoints());
@@ -60,7 +60,7 @@ public class StandaloneCompletedCheckpointStoreTest extends CompletedCheckpointS
@Test
public void testSuspendDiscardsCheckpoints() throws Exception {
CompletedCheckpointStore store = createCompletedCheckpoints(1, ClassLoader.getSystemClassLoader());
- TestCheckpoint checkpoint = createCheckpoint(0);
+ TestCompletedCheckpoint checkpoint = createCheckpoint(0);
store.addCheckpoint(checkpoint);
assertEquals(1, store.getNumberOfRetainedCheckpoints());