You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2018/02/01 15:46:51 UTC
[19/22] flink git commit: [FLINK-8539] [checkpointing] (part 2)
Modify all tests to use CompletedCheckpointStorageLocation.
[FLINK-8539] [checkpointing] (part 2) Modify all tests to use CompletedCheckpointStorageLocation.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/f577d260
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/f577d260
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/f577d260
Branch: refs/heads/master
Commit: f577d2603fb0432fa7a15d54ee25f481bde95d95
Parents: f9dd19b
Author: Stephan Ewen <se...@apache.org>
Authored: Wed Jan 31 16:57:45 2018 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Thu Feb 1 13:54:56 2018 +0100
----------------------------------------------------------------------
.../util/function/FunctionWithException.java | 43 +++++++
.../CheckpointCoordinatorMasterHooksTest.java | 9 +-
.../checkpoint/CheckpointCoordinatorTest.java | 8 +-
.../CheckpointMetadataLoadingTest.java | 13 +-
.../CheckpointSettingsSerializableTest.java | 3 +-
.../checkpoint/CheckpointStateRestoreTest.java | 8 +-
.../CompletedCheckpointStoreTest.java | 16 ++-
.../checkpoint/CompletedCheckpointTest.java | 104 ++++++++-------
.../ZooKeeperCompletedCheckpointStoreTest.java | 8 +-
.../runtime/state/EmptyStreamStateHandle.java | 45 -------
.../AbstractFileCheckpointStorageTestBase.java | 63 +++++----
.../FixFileFsStateOutputStreamTest.java | 127 +++++++++++--------
.../memory/MemoryCheckpointStorageTest.java | 11 +-
.../state/testutils/BackendForTestStream.java | 4 +-
.../state/testutils/EmptyStreamStateHandle.java | 73 +++++++++++
.../TestCompletedCheckpointStorageLocation.java | 66 ++++++++++
.../testingUtils/TestingJobManagerLike.scala | 4 +-
.../StreamTaskStateInitializerImplTest.java | 8 +-
.../tasks/StreamTaskTerminationTest.java | 4 +-
.../streaming/runtime/StateBackendITCase.java | 4 +-
20 files changed, 400 insertions(+), 221 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/f577d260/flink-core/src/main/java/org/apache/flink/util/function/FunctionWithException.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/util/function/FunctionWithException.java b/flink-core/src/main/java/org/apache/flink/util/function/FunctionWithException.java
new file mode 100644
index 0000000..f921e03
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/util/function/FunctionWithException.java
@@ -0,0 +1,43 @@
+/*
+ * 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.util.function;
+
+import org.apache.flink.annotation.Public;
+
+/**
+ * A functional interface for a {@link java.util.function.Function} that may
+ * throw exceptions.
+ *
+ * @param <T> The type of the argument to the function.
+ * @param <R> The type of the result of the supplier.
+ * @param <E> The type of Exceptions thrown by this function.
+ */
+@Public
+@FunctionalInterface
+public interface FunctionWithException<T, R, E extends Throwable> {
+
+ /**
+ * Calls this function.
+ *
+ * @param value The argument to the function.
+ * @return The result of thus supplier.
+ * @throws E This function may throw an exception.
+ */
+ R apply(T value) throws E;
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/f577d260/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorMasterHooksTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorMasterHooksTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorMasterHooksTest.java
index d583a0d..3b345da 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorMasterHooksTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorMasterHooksTest.java
@@ -27,10 +27,10 @@ import org.apache.flink.runtime.executiongraph.ExecutionVertex;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.messages.checkpoint.AcknowledgeCheckpoint;
-import org.apache.flink.runtime.state.EmptyStreamStateHandle;
import org.apache.flink.runtime.state.SharedStateRegistry;
import org.apache.flink.runtime.state.memory.MemoryStateBackend;
+import org.apache.flink.runtime.state.testutils.TestCompletedCheckpointStorageLocation;
import org.junit.Test;
import org.mockito.invocation.InvocationOnMock;
@@ -231,9 +231,7 @@ public class CheckpointCoordinatorMasterHooksTest {
Collections.<OperatorID, OperatorState>emptyMap(),
masterHookStates,
CheckpointProperties.forCheckpoint(CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION),
- new EmptyStreamStateHandle(),
- "pointer");
-
+ new TestCompletedCheckpointStorageLocation());
final ExecutionAttemptID execId = new ExecutionAttemptID();
final ExecutionVertex ackVertex = mockExecutionVertex(execId);
final CheckpointCoordinator cc = instantiateCheckpointCoordinator(jid, ackVertex);
@@ -285,8 +283,7 @@ public class CheckpointCoordinatorMasterHooksTest {
Collections.<OperatorID, OperatorState>emptyMap(),
masterHookStates,
CheckpointProperties.forCheckpoint(CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION),
- new EmptyStreamStateHandle(),
- "pointer");
+ new TestCompletedCheckpointStorageLocation());
final ExecutionAttemptID execId = new ExecutionAttemptID();
final ExecutionVertex ackVertex = mockExecutionVertex(execId);
http://git-wip-us.apache.org/repos/asf/flink/blob/f577d260/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
index 0ca3405..d318575 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
@@ -35,7 +35,6 @@ import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.messages.checkpoint.AcknowledgeCheckpoint;
import org.apache.flink.runtime.messages.checkpoint.DeclineCheckpoint;
import org.apache.flink.runtime.state.ChainedStateHandle;
-import org.apache.flink.runtime.state.EmptyStreamStateHandle;
import org.apache.flink.runtime.state.IncrementalKeyedStateHandle;
import org.apache.flink.runtime.state.KeyGroupRange;
import org.apache.flink.runtime.state.KeyGroupRangeAssignment;
@@ -50,6 +49,7 @@ import org.apache.flink.runtime.state.StreamStateHandle;
import org.apache.flink.runtime.state.filesystem.FileStateHandle;
import org.apache.flink.runtime.state.memory.ByteStreamStateHandle;
import org.apache.flink.runtime.state.memory.MemoryStateBackend;
+import org.apache.flink.runtime.state.testutils.TestCompletedCheckpointStorageLocation;
import org.apache.flink.runtime.testutils.CommonTestUtils;
import org.apache.flink.runtime.testutils.RecoverableCompletedCheckpointStore;
import org.apache.flink.runtime.util.TestByteStreamStateHandleDeepCompare;
@@ -2537,8 +2537,7 @@ public class CheckpointCoordinatorTest extends TestLogger {
operatorStates,
Collections.<MasterState>emptyList(),
CheckpointProperties.forCheckpoint(CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION),
- new EmptyStreamStateHandle(),
- "some/mock/pointer");
+ new TestCompletedCheckpointStorageLocation());
when(standaloneCompletedCheckpointStore.getLatestCheckpoint()).thenReturn(completedCheckpoint);
@@ -3415,8 +3414,7 @@ public class CheckpointCoordinatorTest extends TestLogger {
Collections.<OperatorID, OperatorState>emptyMap(),
Collections.<MasterState>emptyList(),
CheckpointProperties.forCheckpoint(CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION),
- new EmptyStreamStateHandle(),
- "some/dummy/pointer"));
+ new TestCompletedCheckpointStorageLocation()));
CheckpointStatsTracker tracker = mock(CheckpointStatsTracker.class);
coord.setCheckpointStatsTracker(tracker);
http://git-wip-us.apache.org/repos/asf/flink/blob/f577d260/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointMetadataLoadingTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointMetadataLoadingTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointMetadataLoadingTest.java
index 15e6744..70794c6 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointMetadataLoadingTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointMetadataLoadingTest.java
@@ -23,9 +23,11 @@ import org.apache.flink.runtime.checkpoint.savepoint.SavepointV2;
import org.apache.flink.runtime.executiongraph.ExecutionJobVertex;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.state.CompletedCheckpointStorageLocation;
import org.apache.flink.runtime.state.OperatorStateHandle;
import org.apache.flink.runtime.state.StreamStateHandle;
import org.apache.flink.runtime.state.memory.ByteStreamStateHandle;
+import org.apache.flink.runtime.state.testutils.TestCompletedCheckpointStorageLocation;
import org.junit.Rule;
import org.junit.Test;
@@ -91,6 +93,9 @@ public class CheckpointMetadataLoadingTest {
serializedMetadata = new ByteStreamStateHandle("checkpoint", os.toByteArray());
}
+ final CompletedCheckpointStorageLocation storageLocation = new TestCompletedCheckpointStorageLocation(
+ serializedMetadata, "dummy/pointer");
+
ExecutionJobVertex vertex = mock(ExecutionJobVertex.class);
when(vertex.getParallelism()).thenReturn(parallelism);
when(vertex.getMaxParallelism()).thenReturn(parallelism);
@@ -102,7 +107,7 @@ public class CheckpointMetadataLoadingTest {
ClassLoader ucl = Thread.currentThread().getContextClassLoader();
// 1) Load and validate: everything correct
- CompletedCheckpoint loaded = Checkpoints.loadAndValidateCheckpoint(jobId, tasks, "fake/path", serializedMetadata, ucl, false);
+ CompletedCheckpoint loaded = Checkpoints.loadAndValidateCheckpoint(jobId, tasks, storageLocation, ucl, false);
assertEquals(jobId, loaded.getJobId());
assertEquals(checkpointId, loaded.getCheckpointID());
@@ -112,7 +117,7 @@ public class CheckpointMetadataLoadingTest {
when(vertex.isMaxParallelismConfigured()).thenReturn(true);
try {
- Checkpoints.loadAndValidateCheckpoint(jobId, tasks, "fake/path", serializedMetadata, ucl, false);
+ Checkpoints.loadAndValidateCheckpoint(jobId, tasks, storageLocation, ucl, false);
fail("Did not throw expected Exception");
} catch (IllegalStateException expected) {
assertTrue(expected.getMessage().contains("Max parallelism mismatch"));
@@ -122,13 +127,13 @@ public class CheckpointMetadataLoadingTest {
assertNotNull(tasks.remove(jobVertexID));
try {
- Checkpoints.loadAndValidateCheckpoint(jobId, tasks, "fake/path", serializedMetadata, ucl, false);
+ Checkpoints.loadAndValidateCheckpoint(jobId, tasks, storageLocation, ucl, false);
fail("Did not throw expected Exception");
} catch (IllegalStateException expected) {
assertTrue(expected.getMessage().contains("allowNonRestoredState"));
}
// 4) Load and validate: ignore missing vertex
- Checkpoints.loadAndValidateCheckpoint(jobId, tasks, "fake/path", serializedMetadata, ucl, true);
+ Checkpoints.loadAndValidateCheckpoint(jobId, tasks, storageLocation, ucl, true);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/f577d260/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointSettingsSerializableTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointSettingsSerializableTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointSettingsSerializableTest.java
index 479bba2..8abff4f 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointSettingsSerializableTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointSettingsSerializableTest.java
@@ -37,6 +37,7 @@ import org.apache.flink.runtime.jobgraph.tasks.JobCheckpointingSettings;
import org.apache.flink.runtime.query.TaskKvStateRegistry;
import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
import org.apache.flink.runtime.state.CheckpointStorage;
+import org.apache.flink.runtime.state.CompletedCheckpointStorageLocation;
import org.apache.flink.runtime.state.KeyGroupRange;
import org.apache.flink.runtime.state.OperatorStateBackend;
import org.apache.flink.runtime.state.StateBackend;
@@ -149,7 +150,7 @@ public class CheckpointSettingsSerializableTest extends TestLogger {
}
@Override
- public StreamStateHandle resolveCheckpoint(String pointer) throws IOException {
+ public CompletedCheckpointStorageLocation resolveCheckpoint(String pointer) throws IOException {
throw new UnsupportedOperationException();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/f577d260/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointStateRestoreTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointStateRestoreTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointStateRestoreTest.java
index 7348a38..ab353a9 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointStateRestoreTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointStateRestoreTest.java
@@ -28,12 +28,12 @@ import org.apache.flink.runtime.executiongraph.ExecutionVertex;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.messages.checkpoint.AcknowledgeCheckpoint;
-import org.apache.flink.runtime.state.EmptyStreamStateHandle;
import org.apache.flink.runtime.state.KeyGroupRange;
import org.apache.flink.runtime.state.KeyedStateHandle;
import org.apache.flink.runtime.state.OperatorStateHandle;
import org.apache.flink.runtime.state.SharedStateRegistry;
import org.apache.flink.runtime.state.memory.MemoryStateBackend;
+import org.apache.flink.runtime.state.testutils.TestCompletedCheckpointStorageLocation;
import org.apache.flink.util.SerializableObject;
import org.hamcrest.BaseMatcher;
@@ -265,8 +265,7 @@ public class CheckpointStateRestoreTest {
new HashMap<>(checkpointTaskStates),
Collections.<MasterState>emptyList(),
CheckpointProperties.forCheckpoint(CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION),
- new EmptyStreamStateHandle(),
- "<pointer>");
+ new TestCompletedCheckpointStorageLocation());
coord.getCheckpointStore().addCheckpoint(checkpoint);
@@ -293,8 +292,7 @@ public class CheckpointStateRestoreTest {
new HashMap<>(checkpointTaskStates),
Collections.<MasterState>emptyList(),
CheckpointProperties.forCheckpoint(CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION),
- new EmptyStreamStateHandle(),
- "pointer");
+ new TestCompletedCheckpointStorageLocation());
coord.getCheckpointStore().addCheckpoint(checkpoint);
http://git-wip-us.apache.org/repos/asf/flink/blob/f577d260/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 d113aa1..96c95ea 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
@@ -21,9 +21,10 @@ package org.apache.flink.runtime.checkpoint;
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.jobgraph.JobStatus;
import org.apache.flink.runtime.jobgraph.OperatorID;
-import org.apache.flink.runtime.state.EmptyStreamStateHandle;
import org.apache.flink.runtime.state.SharedStateRegistry;
+import org.apache.flink.runtime.state.testutils.TestCompletedCheckpointStorageLocation;
import org.apache.flink.util.TestLogger;
+
import org.junit.Assert;
import org.junit.Test;
@@ -252,13 +253,14 @@ public abstract class CompletedCheckpointStoreTest extends TestLogger {
private transient final CountDownLatch discardLatch = new CountDownLatch(1);
public TestCompletedCheckpoint(
- JobID jobId,
- long checkpointId,
- long timestamp,
- Map<OperatorID, OperatorState> operatorGroupState,
- CheckpointProperties props) {
+ JobID jobId,
+ long checkpointId,
+ long timestamp,
+ Map<OperatorID, OperatorState> operatorGroupState,
+ CheckpointProperties props) {
+
super(jobId, checkpointId, timestamp, Long.MAX_VALUE, operatorGroupState, null, props,
- new EmptyStreamStateHandle(), "<pointer");
+ new TestCompletedCheckpointStorageLocation());
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/f577d260/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointTest.java
index 1bdb947..69003cd 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointTest.java
@@ -19,58 +19,53 @@
package org.apache.flink.runtime.checkpoint;
import org.apache.flink.api.common.JobID;
-import org.apache.flink.core.fs.Path;
import org.apache.flink.core.testutils.CommonTestUtils;
import org.apache.flink.runtime.jobgraph.JobStatus;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.state.SharedStateRegistry;
-import org.apache.flink.runtime.state.StreamStateHandle;
-import org.apache.flink.runtime.state.filesystem.FileStateHandle;
+import org.apache.flink.runtime.state.testutils.EmptyStreamStateHandle;
+import org.apache.flink.runtime.state.testutils.TestCompletedCheckpointStorageLocation;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
-import java.io.File;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
+/**
+ * Unit tests for the {@link CompletedCheckpoint}.
+ */
public class CompletedCheckpointTest {
@Rule
public final TemporaryFolder tmpFolder = new TemporaryFolder();
- /**
- * Tests that persistent checkpoints discard their header file.
- */
@Test
- public void testDiscard() throws Exception {
- File file = tmpFolder.newFile();
- assertEquals(true, file.exists());
-
+ public void testRegisterStatesAtRegistry() {
OperatorState state = mock(OperatorState.class);
- Map<OperatorID, OperatorState> taskStates = new HashMap<>();
- taskStates.put(new OperatorID(), state);
+ Map<OperatorID, OperatorState> operatorStates = new HashMap<>();
+ operatorStates.put(new OperatorID(), state);
- // Verify discard call is forwarded to state
CompletedCheckpoint checkpoint = new CompletedCheckpoint(
new JobID(), 0, 0, 1,
- taskStates,
- Collections.<MasterState>emptyList(),
- CheckpointProperties.forCheckpoint(CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION),
- new FileStateHandle(new Path(file.toURI()), file.length()),
- file.getAbsolutePath());
-
- checkpoint.discardOnShutdown(JobStatus.FAILED);
+ operatorStates,
+ Collections.emptyList(),
+ CheckpointProperties.forCheckpoint(CheckpointRetentionPolicy.RETAIN_ON_FAILURE),
+ new TestCompletedCheckpointStorageLocation());
- assertEquals(false, file.exists());
+ SharedStateRegistry sharedStateRegistry = new SharedStateRegistry();
+ checkpoint.registerSharedStatesAfterRestored(sharedStateRegistry);
+ verify(state, times(1)).registerSharedStates(sharedStateRegistry);
}
/**
@@ -82,18 +77,18 @@ public class CompletedCheckpointTest {
Map<OperatorID, OperatorState> operatorStates = new HashMap<>();
operatorStates.put(new OperatorID(), state);
- StreamStateHandle metadataHandle = mock(StreamStateHandle.class);
+ EmptyStreamStateHandle metadata = new EmptyStreamStateHandle();
+ TestCompletedCheckpointStorageLocation location =
+ new TestCompletedCheckpointStorageLocation(metadata, "ptr");
- boolean discardSubsumed = true;
- CheckpointProperties props = new CheckpointProperties(false, CheckpointType.CHECKPOINT, discardSubsumed, true, true, true, true);
+ CheckpointProperties props = new CheckpointProperties(false, CheckpointType.CHECKPOINT, true, false, false, false, false);
CompletedCheckpoint checkpoint = new CompletedCheckpoint(
new JobID(), 0, 0, 1,
operatorStates,
- Collections.<MasterState>emptyList(),
+ Collections.emptyList(),
props,
- metadataHandle,
- "some/mock/pointer");
+ location);
SharedStateRegistry sharedStateRegistry = new SharedStateRegistry();
checkpoint.registerSharedStatesAfterRestored(sharedStateRegistry);
@@ -103,7 +98,8 @@ public class CompletedCheckpointTest {
checkpoint.discardOnSubsume();
verify(state, times(1)).discardState();
- verify(metadataHandle).discardState();
+ assertTrue(location.isDisposed());
+ assertTrue(metadata.isDisposed());
}
/**
@@ -121,38 +117,45 @@ public class CompletedCheckpointTest {
Map<OperatorID, OperatorState> operatorStates = new HashMap<>();
operatorStates.put(new OperatorID(), state);
- StreamStateHandle metadataHandle = mock(StreamStateHandle.class);
+ EmptyStreamStateHandle retainedHandle = new EmptyStreamStateHandle();
+ TestCompletedCheckpointStorageLocation retainedLocation =
+ new TestCompletedCheckpointStorageLocation(retainedHandle, "ptr");
// Keep
- CheckpointProperties props = new CheckpointProperties(false, CheckpointType.SAVEPOINT, false, false, false, false, false);
+ CheckpointProperties retainProps = new CheckpointProperties(false, CheckpointType.CHECKPOINT, false, false, false, false, false);
CompletedCheckpoint checkpoint = new CompletedCheckpoint(
new JobID(), 0, 0, 1,
new HashMap<>(operatorStates),
- Collections.<MasterState>emptyList(),
- props,
- metadataHandle,
- "mock://some/pointer");
-
- SharedStateRegistry sharedStateRegistry = new SharedStateRegistry();
- checkpoint.registerSharedStatesAfterRestored(sharedStateRegistry);
+ Collections.emptyList(),
+ retainProps,
+ retainedLocation);
checkpoint.discardOnShutdown(status);
+
verify(state, times(0)).discardState();
- verify(metadataHandle, times(0)).discardState();
+ assertFalse(retainedLocation.isDisposed());
+ assertFalse(retainedHandle.isDisposed());
// Discard
- props = new CheckpointProperties(false, CheckpointType.CHECKPOINT, true, true, true, true, true);
+ EmptyStreamStateHandle discardHandle = new EmptyStreamStateHandle();
+ TestCompletedCheckpointStorageLocation discardLocation =
+ new TestCompletedCheckpointStorageLocation(discardHandle, "ptr");
+
+ // Keep
+ CheckpointProperties discardProps = new CheckpointProperties(false, CheckpointType.CHECKPOINT, true, true, true, true, true);
+
checkpoint = new CompletedCheckpoint(
new JobID(), 0, 0, 1,
new HashMap<>(operatorStates),
- Collections.<MasterState>emptyList(),
- props,
- metadataHandle,
- "pointer");
+ Collections.emptyList(),
+ discardProps,
+ discardLocation);
checkpoint.discardOnShutdown(status);
+
verify(state, times(1)).discardState();
- verify(metadataHandle, times(1)).discardState();
+ assertTrue(discardLocation.isDisposed());
+ assertTrue(discardHandle.isDisposed());
}
}
@@ -161,20 +164,15 @@ public class CompletedCheckpointTest {
*/
@Test
public void testCompletedCheckpointStatsCallbacks() throws Exception {
- OperatorState state = mock(OperatorState.class);
- Map<OperatorID, OperatorState> operatorStates = new HashMap<>();
- operatorStates.put(new OperatorID(), state);
-
CompletedCheckpoint completed = new CompletedCheckpoint(
new JobID(),
0,
0,
1,
- new HashMap<>(operatorStates),
- Collections.<MasterState>emptyList(),
+ Collections.emptyMap(),
+ Collections.emptyList(),
CheckpointProperties.forCheckpoint(CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION),
- mock(StreamStateHandle.class),
- "pointer");
+ new TestCompletedCheckpointStorageLocation());
CompletedCheckpointStats.DiscardCallback callback = mock(CompletedCheckpointStats.DiscardCallback.class);
completed.setDiscardCallback(callback);
http://git-wip-us.apache.org/repos/asf/flink/blob/f577d260/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStoreTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStoreTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStoreTest.java
index a167130..a544327 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStoreTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStoreTest.java
@@ -21,8 +21,8 @@ package org.apache.flink.runtime.checkpoint;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.concurrent.Executors;
-import org.apache.flink.runtime.state.EmptyStreamStateHandle;
import org.apache.flink.runtime.state.RetrievableStateHandle;
+import org.apache.flink.runtime.state.testutils.TestCompletedCheckpointStorageLocation;
import org.apache.flink.runtime.zookeeper.RetrievableStateStorageHelper;
import org.apache.flink.runtime.zookeeper.ZooKeeperStateHandleStore;
import org.apache.flink.util.TestLogger;
@@ -96,8 +96,7 @@ public class ZooKeeperCompletedCheckpointStoreTest extends TestLogger {
new HashMap<>(),
null,
CheckpointProperties.forCheckpoint(CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION),
- new EmptyStreamStateHandle(),
- "<pointer>");
+ new TestCompletedCheckpointStorageLocation());
final CompletedCheckpoint completedCheckpoint2 = new CompletedCheckpoint(
new JobID(),
@@ -107,8 +106,7 @@ public class ZooKeeperCompletedCheckpointStoreTest extends TestLogger {
new HashMap<>(),
null,
CheckpointProperties.forCheckpoint(CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION),
- new EmptyStreamStateHandle(),
- "<pointer");
+ new TestCompletedCheckpointStorageLocation());
final Collection<Long> expectedCheckpointIds = new HashSet<>(2);
expectedCheckpointIds.add(1L);
http://git-wip-us.apache.org/repos/asf/flink/blob/f577d260/flink-runtime/src/test/java/org/apache/flink/runtime/state/EmptyStreamStateHandle.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/EmptyStreamStateHandle.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/EmptyStreamStateHandle.java
deleted file mode 100644
index 4b42959..0000000
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/EmptyStreamStateHandle.java
+++ /dev/null
@@ -1,45 +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.state;
-
-import org.apache.flink.core.fs.FSDataInputStream;
-
-import java.io.IOException;
-
-/**
- * A simple dummy implementation of a stream state handle that can be passed in tests.
- * The handle cannot open an input stream.
- */
-public class EmptyStreamStateHandle implements StreamStateHandle {
-
- private static final long serialVersionUID = 0L;
-
- @Override
- public FSDataInputStream openInputStream() throws IOException {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public void discardState() {}
-
- @Override
- public long getStateSize() {
- return 0;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/f577d260/flink-runtime/src/test/java/org/apache/flink/runtime/state/filesystem/AbstractFileCheckpointStorageTestBase.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/filesystem/AbstractFileCheckpointStorageTestBase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/filesystem/AbstractFileCheckpointStorageTestBase.java
index d323ab6..0168ef7 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/filesystem/AbstractFileCheckpointStorageTestBase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/filesystem/AbstractFileCheckpointStorageTestBase.java
@@ -24,9 +24,10 @@ import org.apache.flink.core.fs.FileStatus;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.FileSystem.WriteMode;
import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.state.CheckpointMetadataOutputStream;
import org.apache.flink.runtime.state.CheckpointStorage;
import org.apache.flink.runtime.state.CheckpointStorageLocation;
-import org.apache.flink.runtime.state.CheckpointStreamFactory.CheckpointStateOutputStream;
+import org.apache.flink.runtime.state.CompletedCheckpointStorageLocation;
import org.apache.flink.runtime.state.StreamStateHandle;
import org.junit.Rule;
@@ -74,7 +75,9 @@ public abstract class AbstractFileCheckpointStorageTestBase {
@Test
public void testPointerPathResolution() throws Exception {
final FileSystem fs = FileSystem.getLocalFileSystem();
- final Path metadataFile = new Path(tmp.newFolder().getAbsolutePath(), AbstractFsCheckpointStorage.METADATA_FILE_NAME);
+ final Path metadataFile = new Path(Path.fromLocalFile(tmp.newFolder()), AbstractFsCheckpointStorage.METADATA_FILE_NAME);
+
+ final String basePointer = metadataFile.getParent().toString();
final String pointer1 = metadataFile.toString();
final String pointer2 = metadataFile.getParent().toString();
@@ -89,9 +92,17 @@ public abstract class AbstractFileCheckpointStorageTestBase {
out.write(data);
}
- StreamStateHandle handle1 = storage.resolveCheckpoint(pointer1);
- StreamStateHandle handle2 = storage.resolveCheckpoint(pointer2);
- StreamStateHandle handle3 = storage.resolveCheckpoint(pointer3);
+ CompletedCheckpointStorageLocation completed1 = storage.resolveCheckpoint(pointer1);
+ CompletedCheckpointStorageLocation completed2 = storage.resolveCheckpoint(pointer2);
+ CompletedCheckpointStorageLocation completed3 = storage.resolveCheckpoint(pointer3);
+
+ assertEquals(basePointer, completed1.getExternalPointer());
+ assertEquals(basePointer, completed2.getExternalPointer());
+ assertEquals(basePointer, completed3.getExternalPointer());
+
+ StreamStateHandle handle1 = completed1.getMetadataHandle();
+ StreamStateHandle handle2 = completed2.getMetadataHandle();
+ StreamStateHandle handle3 = completed3.getMetadataHandle();
assertNotNull(handle1);
assertNotNull(handle2);
@@ -156,17 +167,19 @@ public abstract class AbstractFileCheckpointStorageTestBase {
final byte[] data1 = {77, 66, 55, 99, 88};
final byte[] data2 = {1, 3, 2, 5, 4};
- try (CheckpointStateOutputStream out = loc1.createMetadataOutputStream()) {
+ final CompletedCheckpointStorageLocation completedLocation1;
+ try (CheckpointMetadataOutputStream out = loc1.createMetadataOutputStream()) {
out.write(data1);
- out.closeAndGetHandle();
+ completedLocation1 = out.closeAndFinalizeCheckpoint();
}
- final String result1 = loc1.markCheckpointAsFinished();
+ final String result1 = completedLocation1.getExternalPointer();
- try (CheckpointStateOutputStream out = loc2.createMetadataOutputStream()) {
+ final CompletedCheckpointStorageLocation completedLocation2;
+ try (CheckpointMetadataOutputStream out = loc2.createMetadataOutputStream()) {
out.write(data2);
- out.closeAndGetHandle();
+ completedLocation2 = out.closeAndFinalizeCheckpoint();
}
- final String result2 = loc2.markCheckpointAsFinished();
+ final String result2 = completedLocation2.getExternalPointer();
// check that this went to a file, but in a nested directory structure
@@ -184,10 +197,10 @@ public abstract class AbstractFileCheckpointStorageTestBase {
assertTrue(fs.exists(new Path(result2, AbstractFsCheckpointStorage.METADATA_FILE_NAME)));
// check that both storages can resolve each others contents
- validateContents(storage1.resolveCheckpoint(result1), data1);
- validateContents(storage1.resolveCheckpoint(result2), data2);
- validateContents(storage2.resolveCheckpoint(result1), data1);
- validateContents(storage2.resolveCheckpoint(result2), data2);
+ validateContents(storage1.resolveCheckpoint(result1).getMetadataHandle(), data1);
+ validateContents(storage1.resolveCheckpoint(result2).getMetadataHandle(), data2);
+ validateContents(storage2.resolveCheckpoint(result1).getMetadataHandle(), data1);
+ validateContents(storage2.resolveCheckpoint(result2).getMetadataHandle(), data2);
}
@Test
@@ -200,9 +213,9 @@ public abstract class AbstractFileCheckpointStorageTestBase {
// write to the metadata file for the checkpoint
- try (CheckpointStateOutputStream out = loc.createMetadataOutputStream()) {
+ try (CheckpointMetadataOutputStream out = loc.createMetadataOutputStream()) {
out.write(data);
- out.closeAndGetHandle();
+ out.closeAndFinalizeCheckpoint();
}
// create another writer to the metadata file for the checkpoint
@@ -263,15 +276,21 @@ public abstract class AbstractFileCheckpointStorageTestBase {
final byte[] data = {77, 66, 55, 99, 88};
- final StreamStateHandle handle;
- try (CheckpointStateOutputStream out = savepointLocation.createMetadataOutputStream()) {
+ final CompletedCheckpointStorageLocation completed;
+ try (CheckpointMetadataOutputStream out = savepointLocation.createMetadataOutputStream()) {
out.write(data);
- handle = out.closeAndGetHandle();
+ completed = out.closeAndFinalizeCheckpoint();
}
- validateContents(handle, data);
+
+ // we need to do this step to make sure we have a slash (or not) in the same way as the
+ // expected path has it
+ final Path normalizedWithSlash = Path.fromLocalFile(new File(new Path(completed.getExternalPointer()).getParent().getPath()));
+
+ assertEquals(expectedParent, normalizedWithSlash);
+ validateContents(completed.getMetadataHandle(), data);
// validate that the correct directory was used
- FileStateHandle fileStateHandle = (FileStateHandle) handle;
+ FileStateHandle fileStateHandle = (FileStateHandle) completed.getMetadataHandle();
// we need to recreate that path in the same way as the "expected path" (via File and URI) to make
// sure the either both use '/' suffixes, or neither uses them (a bit of an annoying ambiguity)
http://git-wip-us.apache.org/repos/asf/flink/blob/f577d260/flink-runtime/src/test/java/org/apache/flink/runtime/state/filesystem/FixFileFsStateOutputStreamTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/filesystem/FixFileFsStateOutputStreamTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/filesystem/FixFileFsStateOutputStreamTest.java
index 869cc00..7de7f66 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/filesystem/FixFileFsStateOutputStreamTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/filesystem/FixFileFsStateOutputStreamTest.java
@@ -21,18 +21,19 @@ package org.apache.flink.runtime.state.filesystem;
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.FileSystem.WriteMode;
import org.apache.flink.core.fs.Path;
+import org.apache.flink.core.fs.local.LocalDataOutputStream;
+import org.apache.flink.core.fs.local.LocalFileSystem;
import org.apache.flink.core.testutils.CheckedThread;
import org.apache.flink.core.testutils.OneShotLatch;
+import org.apache.flink.util.function.FunctionWithException;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
-import org.mockito.ArgumentCaptor;
-
import java.io.EOFException;
+import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.util.Random;
@@ -43,16 +44,11 @@ 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.anyBoolean;
-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.verify;
-import static org.mockito.Mockito.when;
/**
- * Tests for the {@link }.
+ * Tests for the {@link FixFileFsStateOutputStream}.
*/
public class FixFileFsStateOutputStreamTest {
@@ -69,41 +65,46 @@ public class FixFileFsStateOutputStreamTest {
@Test
public void testEmptyState() throws Exception {
final FileSystem fs = FileSystem.getLocalFileSystem();
- final Path path = new Path(new Path(tmp.newFolder().toURI()), "myFileName");
- final FileStateHandle handle;
- try (FixFileFsStateOutputStream stream = new FixFileFsStateOutputStream(fs, path)) {
- handle = stream.closeAndGetHandle();
+ final Path checkpointDir = Path.fromLocalFile(tmp.newFolder());
+ final Path metadataPath = new Path(checkpointDir, "myFileName");
+
+ final FsCompletedCheckpointStorageLocation location;
+ try (FixFileFsStateOutputStream stream = new FixFileFsStateOutputStream(fs, metadataPath, checkpointDir)) {
+ location = stream.closeAndFinalizeCheckpoint();
}
// must have created a handle
- assertNotNull(handle);
- assertEquals(path, handle.getFilePath());
+ assertNotNull(location);
+ assertNotNull(location.getMetadataHandle());
+ assertEquals(metadataPath, location.getMetadataHandle().getFilePath());
- // the pointer path should exist as a directory
- assertTrue(fs.exists(handle.getFilePath()));
- assertFalse(fs.getFileStatus(path).isDir());
+ // the pointer path should exist as a file
+ assertTrue(fs.exists(metadataPath));
+ assertFalse(fs.getFileStatus(metadataPath).isDir());
// the contents should be empty
- try (FSDataInputStream in = handle.openInputStream()) {
+ try (FSDataInputStream in = location.getMetadataHandle().openInputStream()) {
assertEquals(-1, in.read());
}
}
/**
- * Simple write and read test
+ * Simple write and read test.
*/
@Test
public void testWriteAndRead() throws Exception {
final FileSystem fs = FileSystem.getLocalFileSystem();
- final Path path = new Path(new Path(tmp.newFolder().toURI()), "fooBarName");
+
+ final Path checkpointDir = Path.fromLocalFile(tmp.newFolder());
+ final Path metadataPath = new Path(checkpointDir, "fooBarName");
final Random rnd = new Random();
final byte[] data = new byte[1694523];
// write the data (mixed single byte writes and array writes)
- final FileStateHandle handle;
- try (FixFileFsStateOutputStream stream = new FixFileFsStateOutputStream(fs, path)) {
+ final FsCompletedCheckpointStorageLocation completed;
+ try (FixFileFsStateOutputStream stream = new FixFileFsStateOutputStream(fs, metadataPath, checkpointDir)) {
for (int i = 0; i < data.length;) {
if (rnd.nextBoolean()) {
stream.write(data[i++]);
@@ -114,18 +115,18 @@ public class FixFileFsStateOutputStreamTest {
i += len;
}
}
- handle = stream.closeAndGetHandle();
+ completed = stream.closeAndFinalizeCheckpoint();
}
// (1) stream from handle must hold the contents
- try (FSDataInputStream in = handle.openInputStream()) {
+ try (FSDataInputStream in = completed.getMetadataHandle().openInputStream()) {
byte[] buffer = new byte[data.length];
readFully(in, buffer);
assertArrayEquals(data, buffer);
}
// (2) the pointer must point to a file with that contents
- try (FSDataInputStream in = fs.open(handle.getFilePath())) {
+ try (FSDataInputStream in = fs.open(completed.getMetadataHandle().getFilePath())) {
byte[] buffer = new byte[data.length];
readFully(in, buffer);
assertArrayEquals(data, buffer);
@@ -138,18 +139,21 @@ public class FixFileFsStateOutputStreamTest {
@Test
public void testCleanupWhenClosingStream() throws IOException {
final FileSystem fs = FileSystem.getLocalFileSystem();
- final Path path = new Path(new Path(tmp.newFolder().toURI()), "nonCreativeTestFileName");
+
+ final Path checkpointDir = Path.fromLocalFile(tmp.newFolder());
+ final Path metadataPath = new Path(checkpointDir, "nonCreativeTestFileName");
// write some test data and close the stream
- try (FixFileFsStateOutputStream stream = new FixFileFsStateOutputStream(fs, path)) {
+ try (FixFileFsStateOutputStream stream = new FixFileFsStateOutputStream(fs, metadataPath, checkpointDir)) {
Random rnd = new Random();
for (int i = 0; i < rnd.nextInt(1000); i++) {
stream.write(rnd.nextInt(100));
}
- assertTrue(fs.exists(path));
+ assertTrue(fs.exists(metadataPath));
}
- assertFalse(fs.exists(path));
+ assertFalse(fs.exists(metadataPath));
+ assertTrue(fs.exists(checkpointDir));
}
/**
@@ -157,30 +161,23 @@ public class FixFileFsStateOutputStreamTest {
*/
@Test
public void testCleanupWhenFailingCloseAndGetHandle() throws IOException {
- final Path path = new Path(new Path(tmp.newFolder().toURI()), "neverCreated");
-
- final FileSystem fs = mock(FileSystem.class);
- final FSDataOutputStream outputStream = mock(FSDataOutputStream.class);
-
- final ArgumentCaptor<Path> pathCaptor = ArgumentCaptor.forClass(Path.class);
-
- when(fs.create(pathCaptor.capture(), any(WriteMode.class))).thenReturn(outputStream);
- doThrow(new IOException("Test IOException.")).when(outputStream).close();
+ final Path checkpointDir = Path.fromLocalFile(tmp.newFolder());
+ final Path metadataPath = new Path(checkpointDir, "test_name");
- FixFileFsStateOutputStream stream = new FixFileFsStateOutputStream(fs, path);
- verify(fs).create(any(Path.class), any(WriteMode.class));
+ final FileSystem fs = spy(new TestFs((path) -> new FailingCloseStream(new File(path.getPath()))));
- stream.write(new byte[] {1,2,3,4,5});
+ FixFileFsStateOutputStream stream = new FixFileFsStateOutputStream(fs, metadataPath, checkpointDir);
+ stream.write(new byte[] {1, 2, 3, 4, 5});
try {
- stream.closeAndGetHandle();
+ stream.closeAndFinalizeCheckpoint();
fail("Expected IOException");
}
catch (IOException ignored) {
// expected exception
}
- verify(fs).delete(eq(pathCaptor.getValue()), anyBoolean());
+ verify(fs).delete(metadataPath, false);
}
/**
@@ -191,15 +188,13 @@ public class FixFileFsStateOutputStreamTest {
*/
@Test
public void testCloseDoesNotLock() throws Exception {
- // a stream that blocks but is released when closed
- final FSDataOutputStream stream = new BlockerStream();
+ final Path checkpointDir = Path.fromLocalFile(tmp.newFolder());
+ final Path metadataPath = new Path(checkpointDir, "this-is-ignored-anyways.file");
- final FileSystem fileSystem = mock(FileSystem.class);
- when(fileSystem.create(any(Path.class), any(WriteMode.class))).thenReturn(stream);
+ final FileSystem fileSystem = spy(new TestFs((path) -> new BlockerStream()));
- final Path path = new Path(new Path(tmp.newFolder().toURI()), "this-is-ignored-anyways.file");
final FixFileFsStateOutputStream checkpointStream =
- new FixFileFsStateOutputStream(fileSystem, path);
+ new FixFileFsStateOutputStream(fileSystem, metadataPath, checkpointDir);
final OneShotLatch sync = new OneShotLatch();
@@ -209,7 +204,7 @@ public class FixFileFsStateOutputStreamTest {
public void go() throws Exception {
sync.trigger();
// that call should now block, because it accesses the position
- checkpointStream.closeAndGetHandle();
+ checkpointStream.closeAndFinalizeCheckpoint();
}
};
thread.start();
@@ -282,4 +277,32 @@ public class FixFileFsStateOutputStreamTest {
throw new IOException("closed");
}
}
+
+ // ------------------------------------------------------------------------
+
+ private static class FailingCloseStream extends LocalDataOutputStream {
+
+ FailingCloseStream(File file) throws IOException {
+ super(file);
+ }
+
+ @Override
+ public void close() throws IOException {
+ throw new IOException();
+ }
+ }
+
+ private static class TestFs extends LocalFileSystem {
+
+ private final FunctionWithException<Path, FSDataOutputStream, IOException> streamFactory;
+
+ TestFs(FunctionWithException<Path, FSDataOutputStream, IOException> streamFactory) {
+ this.streamFactory = streamFactory;
+ }
+
+ @Override
+ public FSDataOutputStream create(Path filePath, WriteMode overwrite) throws IOException {
+ return streamFactory.apply(filePath);
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/f577d260/flink-runtime/src/test/java/org/apache/flink/runtime/state/memory/MemoryCheckpointStorageTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/memory/MemoryCheckpointStorageTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/memory/MemoryCheckpointStorageTest.java
index 8e64884..7e1cf77 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/memory/MemoryCheckpointStorageTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/memory/MemoryCheckpointStorageTest.java
@@ -20,9 +20,11 @@ package org.apache.flink.runtime.state.memory;
import org.apache.flink.api.common.JobID;
import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.state.CheckpointMetadataOutputStream;
import org.apache.flink.runtime.state.CheckpointStorage;
import org.apache.flink.runtime.state.CheckpointStorageLocation;
import org.apache.flink.runtime.state.CheckpointStreamFactory.CheckpointStateOutputStream;
+import org.apache.flink.runtime.state.CompletedCheckpointStorageLocation;
import org.apache.flink.runtime.state.StreamStateHandle;
import org.apache.flink.runtime.state.filesystem.AbstractFileCheckpointStorageTestBase;
import org.apache.flink.runtime.state.memory.MemCheckpointStreamFactory.MemoryCheckpointOutputStream;
@@ -119,15 +121,16 @@ public class MemoryCheckpointStorageTest extends AbstractFileCheckpointStorageTe
CheckpointStorageLocation location = storage.initializeLocationForCheckpoint(9);
- CheckpointStateOutputStream stream = location.createMetadataOutputStream();
+ CheckpointMetadataOutputStream stream = location.createMetadataOutputStream();
stream.write(99);
- StreamStateHandle handle = stream.closeAndGetHandle();
+
+ CompletedCheckpointStorageLocation completed = stream.closeAndFinalizeCheckpoint();
+ StreamStateHandle handle = completed.getMetadataHandle();
assertTrue(handle instanceof ByteStreamStateHandle);
// the reference is not valid in that case
- String reference = location.markCheckpointAsFinished();
try {
- storage.resolveCheckpoint(reference);
+ storage.resolveCheckpoint(completed.getExternalPointer());
fail("should fail with an exception");
} catch (Exception e) {
// expected
http://git-wip-us.apache.org/repos/asf/flink/blob/f577d260/flink-runtime/src/test/java/org/apache/flink/runtime/state/testutils/BackendForTestStream.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/testutils/BackendForTestStream.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/testutils/BackendForTestStream.java
index bca0a48..b40f179 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/testutils/BackendForTestStream.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/testutils/BackendForTestStream.java
@@ -26,7 +26,7 @@ import org.apache.flink.runtime.state.CheckpointStorageLocationReference;
import org.apache.flink.runtime.state.CheckpointStreamFactory;
import org.apache.flink.runtime.state.CheckpointStreamFactory.CheckpointStateOutputStream;
import org.apache.flink.runtime.state.CheckpointedStateScope;
-import org.apache.flink.runtime.state.StreamStateHandle;
+import org.apache.flink.runtime.state.CompletedCheckpointStorageLocation;
import org.apache.flink.runtime.state.memory.MemoryStateBackend;
import org.apache.flink.util.function.SupplierWithException;
@@ -85,7 +85,7 @@ public class BackendForTestStream extends MemoryStateBackend {
}
@Override
- public StreamStateHandle resolveCheckpoint(String pointer) throws IOException {
+ public CompletedCheckpointStorageLocation resolveCheckpoint(String pointer) throws IOException {
throw new UnsupportedOperationException();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/f577d260/flink-runtime/src/test/java/org/apache/flink/runtime/state/testutils/EmptyStreamStateHandle.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/testutils/EmptyStreamStateHandle.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/testutils/EmptyStreamStateHandle.java
new file mode 100644
index 0000000..3dad010
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/testutils/EmptyStreamStateHandle.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.state.testutils;
+
+import org.apache.flink.core.fs.FSDataInputStream;
+import org.apache.flink.runtime.state.StreamStateHandle;
+
+import java.io.IOException;
+
+/**
+ * A simple dummy implementation of a stream state handle that can be passed in tests.
+ * The handle cannot open an input stream.
+ */
+public class EmptyStreamStateHandle implements StreamStateHandle {
+
+ private static final long serialVersionUID = 0L;
+
+ private boolean disposed;
+
+ public boolean isDisposed() {
+ return disposed;
+ }
+
+ @Override
+ public FSDataInputStream openInputStream() throws IOException {
+ // returns an empty stream
+ return new FSDataInputStream() {
+
+ @Override
+ public void seek(long desired) throws IOException {
+ if (desired != 0) {
+ throw new IOException("out of bounds");
+ }
+ }
+
+ @Override
+ public long getPos() {
+ return 0;
+ }
+
+ @Override
+ public int read() throws IOException {
+ return -1;
+ }
+ };
+ }
+
+ @Override
+ public void discardState() throws Exception {
+ disposed = true;
+ }
+
+ @Override
+ public long getStateSize() {
+ return 0;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/f577d260/flink-runtime/src/test/java/org/apache/flink/runtime/state/testutils/TestCompletedCheckpointStorageLocation.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/testutils/TestCompletedCheckpointStorageLocation.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/testutils/TestCompletedCheckpointStorageLocation.java
new file mode 100644
index 0000000..dada952
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/testutils/TestCompletedCheckpointStorageLocation.java
@@ -0,0 +1,66 @@
+/*
+ * 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.state.testutils;
+
+import org.apache.flink.runtime.state.CompletedCheckpointStorageLocation;
+import org.apache.flink.runtime.state.StreamStateHandle;
+
+import java.io.IOException;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * A {@link CompletedCheckpointStorageLocation} for tests.
+ */
+public class TestCompletedCheckpointStorageLocation implements CompletedCheckpointStorageLocation {
+
+ private final StreamStateHandle metadataHandle;
+
+ private final String pointer;
+
+ private boolean disposed;
+
+ public TestCompletedCheckpointStorageLocation() {
+ this(new EmptyStreamStateHandle(), "<pointer>");
+ }
+
+ public TestCompletedCheckpointStorageLocation(StreamStateHandle metadataHandle, String pointer) {
+ this.metadataHandle = checkNotNull(metadataHandle);
+ this.pointer = checkNotNull(pointer);
+ }
+
+ public boolean isDisposed() {
+ return disposed;
+ }
+
+ @Override
+ public String getExternalPointer() {
+ return pointer;
+ }
+
+ @Override
+ public StreamStateHandle getMetadataHandle() {
+ return metadataHandle;
+ }
+
+ @Override
+ public void disposeStorageLocation() throws IOException {
+ disposed = true;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/f577d260/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManagerLike.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManagerLike.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManagerLike.scala
index 7cf466b..2c49d0f 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManagerLike.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManagerLike.scala
@@ -323,9 +323,9 @@ trait TestingJobManagerLike extends FlinkActor {
flinkConfiguration, classloader, null)
val backend = if (loadedBackend != null) loadedBackend else new MemoryStateBackend()
- val metadataHandle = backend.resolveCheckpoint(savepointPath)
+ val checkpointLocation = backend.resolveCheckpoint(savepointPath)
- val stream = new DataInputStream(metadataHandle.openInputStream())
+ val stream = new DataInputStream(checkpointLocation.getMetadataHandle.openInputStream())
val savepoint = try {
Checkpoints.loadCheckpointMetadata(stream, classloader)
}
http://git-wip-us.apache.org/repos/asf/flink/blob/f577d260/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImplTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImplTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImplTest.java
index 6b69707..0eb140a 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImplTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImplTest.java
@@ -33,13 +33,13 @@ import org.apache.flink.runtime.operators.testutils.DummyEnvironment;
import org.apache.flink.runtime.query.TaskKvStateRegistry;
import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
import org.apache.flink.runtime.state.CheckpointStorage;
+import org.apache.flink.runtime.state.CompletedCheckpointStorageLocation;
import org.apache.flink.runtime.state.KeyGroupRange;
import org.apache.flink.runtime.state.KeyGroupStatePartitionStreamProvider;
import org.apache.flink.runtime.state.OperatorStateBackend;
import org.apache.flink.runtime.state.OperatorStateHandle;
import org.apache.flink.runtime.state.StateBackend;
import org.apache.flink.runtime.state.StatePartitionStreamProvider;
-import org.apache.flink.runtime.state.StreamStateHandle;
import org.apache.flink.runtime.state.TaskStateManager;
import org.apache.flink.runtime.state.TaskStateManagerImplTest;
import org.apache.flink.runtime.state.memory.MemoryStateBackend;
@@ -139,13 +139,13 @@ public class StreamTaskStateInitializerImplTest {
StateBackend mockingBackend = spy(new StateBackend() {
@Override
- public StreamStateHandle resolveCheckpoint(String pointer) throws IOException {
- return null;
+ public CompletedCheckpointStorageLocation resolveCheckpoint(String pointer) throws IOException {
+ throw new UnsupportedOperationException();
}
@Override
public CheckpointStorage createCheckpointStorage(JobID jobId) throws IOException {
- return null;
+ throw new UnsupportedOperationException();
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/f577d260/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTerminationTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTerminationTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTerminationTest.java
index f88857b..24b2014 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTerminationTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTerminationTest.java
@@ -54,11 +54,11 @@ import org.apache.flink.runtime.query.TaskKvStateRegistry;
import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
import org.apache.flink.runtime.state.CheckpointStorage;
import org.apache.flink.runtime.state.CheckpointStreamFactory;
+import org.apache.flink.runtime.state.CompletedCheckpointStorageLocation;
import org.apache.flink.runtime.state.KeyGroupRange;
import org.apache.flink.runtime.state.OperatorStateBackend;
import org.apache.flink.runtime.state.OperatorStateHandle;
import org.apache.flink.runtime.state.StateBackend;
-import org.apache.flink.runtime.state.StreamStateHandle;
import org.apache.flink.runtime.state.TestTaskStateManager;
import org.apache.flink.runtime.state.memory.MemoryBackendCheckpointStorage;
import org.apache.flink.runtime.taskmanager.CheckpointResponder;
@@ -251,7 +251,7 @@ public class StreamTaskTerminationTest extends TestLogger {
private static final long serialVersionUID = -5053068148933314100L;
@Override
- public StreamStateHandle resolveCheckpoint(String pointer) throws IOException {
+ public CompletedCheckpointStorageLocation resolveCheckpoint(String pointer) throws IOException {
throw new UnsupportedOperationException();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/f577d260/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/StateBackendITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/StateBackendITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/StateBackendITCase.java
index df8d40e..46c408e 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/StateBackendITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/StateBackendITCase.java
@@ -30,10 +30,10 @@ import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.runtime.query.TaskKvStateRegistry;
import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
import org.apache.flink.runtime.state.CheckpointStorage;
+import org.apache.flink.runtime.state.CompletedCheckpointStorageLocation;
import org.apache.flink.runtime.state.KeyGroupRange;
import org.apache.flink.runtime.state.OperatorStateBackend;
import org.apache.flink.runtime.state.StateBackend;
-import org.apache.flink.runtime.state.StreamStateHandle;
import org.apache.flink.runtime.state.memory.MemoryBackendCheckpointStorage;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.test.util.AbstractTestBase;
@@ -94,7 +94,7 @@ public class StateBackendITCase extends AbstractTestBase {
private static final long serialVersionUID = 1L;
@Override
- public StreamStateHandle resolveCheckpoint(String pointer) throws IOException {
+ public CompletedCheckpointStorageLocation resolveCheckpoint(String pointer) throws IOException {
throw new UnsupportedOperationException();
}