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:50 UTC
[18/22] flink git commit: [FLINK-8539] [checkpointing] (part 3)
Rename FixFileFsStateOutputStream to FsCheckpointMetadataOutputStream
[FLINK-8539] [checkpointing] (part 3) Rename FixFileFsStateOutputStream to FsCheckpointMetadataOutputStream
The new name captures the proper use and meaning of the class in a better way.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/d85a62db
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/d85a62db
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/d85a62db
Branch: refs/heads/master
Commit: d85a62db13a1d90159e7e8f924c8006febad552b
Parents: f577d26
Author: Stephan Ewen <se...@apache.org>
Authored: Wed Jan 31 17:01:33 2018 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Thu Feb 1 13:54:56 2018 +0100
----------------------------------------------------------------------
.../filesystem/FixFileFsStateOutputStream.java | 164 ----------
.../FsCheckpointMetadataOutputStream.java | 159 ++++++++++
.../filesystem/FsCheckpointStorageLocation.java | 2 +-
...istentMetadataCheckpointStorageLocation.java | 4 +-
.../FixFileFsStateOutputStreamTest.java | 308 -------------------
.../FsCheckpointMetadataOutputStreamTest.java | 308 +++++++++++++++++++
6 files changed, 470 insertions(+), 475 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/d85a62db/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FixFileFsStateOutputStream.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FixFileFsStateOutputStream.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FixFileFsStateOutputStream.java
deleted file mode 100644
index 046863b..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FixFileFsStateOutputStream.java
+++ /dev/null
@@ -1,164 +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.filesystem;
-
-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.runtime.state.CheckpointMetadataOutputStream;
-import org.apache.flink.runtime.state.CheckpointStreamFactory.CheckpointStateOutputStream;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import javax.annotation.Nonnull;
-
-import java.io.IOException;
-
-import static org.apache.flink.util.Preconditions.checkNotNull;
-
-/**
- * A {@link CheckpointStateOutputStream} that writes into a specified file and
- * returns a {@link FileStateHandle} upon closing.
- *
- * <p>Unlike the {@link org.apache.flink.runtime.state.filesystem.FsCheckpointStreamFactory.FsCheckpointStateOutputStream},
- * this stream does not have a threshold below which it returns a memory byte stream handle,
- * and does not create random files, but writes to a specified file.
- */
-public final class FixFileFsStateOutputStream extends CheckpointMetadataOutputStream {
-
- private static final Logger LOG = LoggerFactory.getLogger(FixFileFsStateOutputStream.class);
-
- // ------------------------------------------------------------------------
-
- private final FSDataOutputStream out;
-
- private final Path metadataFilePath;
-
- private final Path exclusiveCheckpointDir;
-
- private final FileSystem fileSystem;
-
- private volatile boolean closed;
-
- public FixFileFsStateOutputStream(
- FileSystem fileSystem,
- Path metadataFilePath,
- Path exclusiveCheckpointDir) throws IOException {
-
- this.fileSystem = checkNotNull(fileSystem);
- this.metadataFilePath = checkNotNull(metadataFilePath);
- this.exclusiveCheckpointDir = checkNotNull(exclusiveCheckpointDir);
-
- this.out = fileSystem.create(metadataFilePath, WriteMode.NO_OVERWRITE);
- }
-
- // ------------------------------------------------------------------------
- // I/O
- // ------------------------------------------------------------------------
-
- @Override
- public final void write(int b) throws IOException {
- out.write(b);
- }
-
- @Override
- public final void write(@Nonnull byte[] b, int off, int len) throws IOException {
- out.write(b, off, len);
- }
-
- @Override
- public long getPos() throws IOException {
- return out.getPos();
- }
-
- @Override
- public void flush() throws IOException {
- out.flush();
- }
-
- @Override
- public void sync() throws IOException {
- out.sync();
- }
-
- // ------------------------------------------------------------------------
- // Closing
- // ------------------------------------------------------------------------
-
- public boolean isClosed() {
- return closed;
- }
-
- @Override
- public void close() {
- if (!closed) {
- closed = true;
-
- try {
- out.close();
- fileSystem.delete(metadataFilePath, false);
- }
- catch (Throwable t) {
- LOG.warn("Could not close the state stream for {}.", metadataFilePath, t);
- }
- }
- }
-
- @Override
- public FsCompletedCheckpointStorageLocation closeAndFinalizeCheckpoint() throws IOException {
- synchronized (this) {
- if (!closed) {
- try {
- // make a best effort attempt to figure out the size
- long size = 0;
- try {
- size = out.getPos();
- } catch (Exception ignored) {}
-
- out.close();
-
- FileStateHandle metaDataHandle = new FileStateHandle(metadataFilePath, size);
-
- return new FsCompletedCheckpointStorageLocation(
- fileSystem, exclusiveCheckpointDir, metaDataHandle, exclusiveCheckpointDir.toString());
- }
- catch (Exception e) {
- try {
- fileSystem.delete(metadataFilePath, false);
- }
- catch (Exception deleteException) {
- LOG.warn("Could not delete the checkpoint stream file {}.", metadataFilePath, deleteException);
- }
-
- throw new IOException("Could not flush and close the file system " +
- "output stream to " + metadataFilePath + " in order to obtain the " +
- "stream state handle", e);
- }
- finally {
- closed = true;
- }
- }
- else {
- throw new IOException("Stream has already been closed and discarded.");
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/d85a62db/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsCheckpointMetadataOutputStream.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsCheckpointMetadataOutputStream.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsCheckpointMetadataOutputStream.java
new file mode 100644
index 0000000..f18d578
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsCheckpointMetadataOutputStream.java
@@ -0,0 +1,159 @@
+/*
+ * 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.filesystem;
+
+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.runtime.state.CheckpointMetadataOutputStream;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nonnull;
+
+import java.io.IOException;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * A {@link CheckpointMetadataOutputStream} that writes a specified file and directory, and
+ * returns a {@link FsCompletedCheckpointStorageLocation} upon closing.
+ */
+public final class FsCheckpointMetadataOutputStream extends CheckpointMetadataOutputStream {
+
+ private static final Logger LOG = LoggerFactory.getLogger(FsCheckpointMetadataOutputStream.class);
+
+ // ------------------------------------------------------------------------
+
+ private final FSDataOutputStream out;
+
+ private final Path metadataFilePath;
+
+ private final Path exclusiveCheckpointDir;
+
+ private final FileSystem fileSystem;
+
+ private volatile boolean closed;
+
+ public FsCheckpointMetadataOutputStream(
+ FileSystem fileSystem,
+ Path metadataFilePath,
+ Path exclusiveCheckpointDir) throws IOException {
+
+ this.fileSystem = checkNotNull(fileSystem);
+ this.metadataFilePath = checkNotNull(metadataFilePath);
+ this.exclusiveCheckpointDir = checkNotNull(exclusiveCheckpointDir);
+
+ this.out = fileSystem.create(metadataFilePath, WriteMode.NO_OVERWRITE);
+ }
+
+ // ------------------------------------------------------------------------
+ // I/O
+ // ------------------------------------------------------------------------
+
+ @Override
+ public final void write(int b) throws IOException {
+ out.write(b);
+ }
+
+ @Override
+ public final void write(@Nonnull byte[] b, int off, int len) throws IOException {
+ out.write(b, off, len);
+ }
+
+ @Override
+ public long getPos() throws IOException {
+ return out.getPos();
+ }
+
+ @Override
+ public void flush() throws IOException {
+ out.flush();
+ }
+
+ @Override
+ public void sync() throws IOException {
+ out.sync();
+ }
+
+ // ------------------------------------------------------------------------
+ // Closing
+ // ------------------------------------------------------------------------
+
+ public boolean isClosed() {
+ return closed;
+ }
+
+ @Override
+ public void close() {
+ if (!closed) {
+ closed = true;
+
+ try {
+ out.close();
+ fileSystem.delete(metadataFilePath, false);
+ }
+ catch (Throwable t) {
+ LOG.warn("Could not close the state stream for {}.", metadataFilePath, t);
+ }
+ }
+ }
+
+ @Override
+ public FsCompletedCheckpointStorageLocation closeAndFinalizeCheckpoint() throws IOException {
+ synchronized (this) {
+ if (!closed) {
+ try {
+ // make a best effort attempt to figure out the size
+ long size = 0;
+ try {
+ size = out.getPos();
+ } catch (Exception ignored) {}
+
+ out.close();
+
+ FileStateHandle metaDataHandle = new FileStateHandle(metadataFilePath, size);
+
+ return new FsCompletedCheckpointStorageLocation(
+ fileSystem, exclusiveCheckpointDir, metaDataHandle, exclusiveCheckpointDir.toString());
+ }
+ catch (Exception e) {
+ try {
+ fileSystem.delete(metadataFilePath, false);
+ }
+ catch (Exception deleteException) {
+ LOG.warn("Could not delete the checkpoint stream file {}.", metadataFilePath, deleteException);
+ }
+
+ throw new IOException("Could not flush and close the file system " +
+ "output stream to " + metadataFilePath + " in order to obtain the " +
+ "stream state handle", e);
+ }
+ finally {
+ closed = true;
+ }
+ }
+ else {
+ throw new IOException("Stream has already been closed and discarded.");
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/d85a62db/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStorageLocation.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStorageLocation.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStorageLocation.java
index 48d0e8b..469507a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStorageLocation.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStorageLocation.java
@@ -96,7 +96,7 @@ public class FsCheckpointStorageLocation extends FsCheckpointStreamFactory imple
@Override
public CheckpointMetadataOutputStream createMetadataOutputStream() throws IOException {
- return new FixFileFsStateOutputStream(fileSystem, metadataFilePath, checkpointDirectory);
+ return new FsCheckpointMetadataOutputStream(fileSystem, metadataFilePath, checkpointDirectory);
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/d85a62db/flink-runtime/src/main/java/org/apache/flink/runtime/state/memory/PersistentMetadataCheckpointStorageLocation.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/memory/PersistentMetadataCheckpointStorageLocation.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/memory/PersistentMetadataCheckpointStorageLocation.java
index c9b5068..e8eb607 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/memory/PersistentMetadataCheckpointStorageLocation.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/memory/PersistentMetadataCheckpointStorageLocation.java
@@ -24,7 +24,7 @@ import org.apache.flink.runtime.state.CheckpointMetadataOutputStream;
import org.apache.flink.runtime.state.CheckpointStorageLocation;
import org.apache.flink.runtime.state.CheckpointStorageLocationReference;
import org.apache.flink.runtime.state.filesystem.AbstractFsCheckpointStorage;
-import org.apache.flink.runtime.state.filesystem.FixFileFsStateOutputStream;
+import org.apache.flink.runtime.state.filesystem.FsCheckpointMetadataOutputStream;
import java.io.IOException;
@@ -67,7 +67,7 @@ public class PersistentMetadataCheckpointStorageLocation
@Override
public CheckpointMetadataOutputStream createMetadataOutputStream() throws IOException {
- return new FixFileFsStateOutputStream(fileSystem, metadataFilePath, checkpointDirectory);
+ return new FsCheckpointMetadataOutputStream(fileSystem, metadataFilePath, checkpointDirectory);
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/d85a62db/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
deleted file mode 100644
index 7de7f66..0000000
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/filesystem/FixFileFsStateOutputStreamTest.java
+++ /dev/null
@@ -1,308 +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.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.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 java.io.EOFException;
-import java.io.File;
-import java.io.IOException;
-import java.io.InputStream;
-import java.util.Random;
-
-import static org.junit.Assert.assertArrayEquals;
-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.Mockito.spy;
-import static org.mockito.Mockito.verify;
-
-/**
- * Tests for the {@link FixFileFsStateOutputStream}.
- */
-public class FixFileFsStateOutputStreamTest {
-
- @Rule
- public final TemporaryFolder tmp = new TemporaryFolder();
-
- // ------------------------------------------------------------------------
- // Tests
- // ------------------------------------------------------------------------
-
- /**
- * Validates that even empty streams create a file and a file state handle.
- */
- @Test
- public void testEmptyState() throws Exception {
- final FileSystem fs = FileSystem.getLocalFileSystem();
-
- 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(location);
- assertNotNull(location.getMetadataHandle());
- assertEquals(metadataPath, location.getMetadataHandle().getFilePath());
-
- // 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 = location.getMetadataHandle().openInputStream()) {
- assertEquals(-1, in.read());
- }
- }
-
- /**
- * Simple write and read test.
- */
- @Test
- public void testWriteAndRead() throws Exception {
- final FileSystem fs = FileSystem.getLocalFileSystem();
-
- 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 FsCompletedCheckpointStorageLocation completed;
- try (FixFileFsStateOutputStream stream = new FixFileFsStateOutputStream(fs, metadataPath, checkpointDir)) {
- for (int i = 0; i < data.length;) {
- if (rnd.nextBoolean()) {
- stream.write(data[i++]);
- }
- else {
- int len = rnd.nextInt(Math.min(data.length - i, 32));
- stream.write(data, i, len);
- i += len;
- }
- }
- completed = stream.closeAndFinalizeCheckpoint();
- }
-
- // (1) stream from handle must hold the contents
- 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(completed.getMetadataHandle().getFilePath())) {
- byte[] buffer = new byte[data.length];
- readFully(in, buffer);
- assertArrayEquals(data, buffer);
- }
- }
-
- /**
- * Tests that the underlying stream file is deleted upon calling close.
- */
- @Test
- public void testCleanupWhenClosingStream() throws IOException {
- final FileSystem fs = FileSystem.getLocalFileSystem();
-
- 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, metadataPath, checkpointDir)) {
- Random rnd = new Random();
- for (int i = 0; i < rnd.nextInt(1000); i++) {
- stream.write(rnd.nextInt(100));
- }
- assertTrue(fs.exists(metadataPath));
- }
-
- assertFalse(fs.exists(metadataPath));
- assertTrue(fs.exists(checkpointDir));
- }
-
- /**
- * Tests that the underlying stream file is deleted if the closeAndGetHandle method fails.
- */
- @Test
- public void testCleanupWhenFailingCloseAndGetHandle() throws IOException {
- final Path checkpointDir = Path.fromLocalFile(tmp.newFolder());
- final Path metadataPath = new Path(checkpointDir, "test_name");
-
- final FileSystem fs = spy(new TestFs((path) -> new FailingCloseStream(new File(path.getPath()))));
-
- FixFileFsStateOutputStream stream = new FixFileFsStateOutputStream(fs, metadataPath, checkpointDir);
- stream.write(new byte[] {1, 2, 3, 4, 5});
-
- try {
- stream.closeAndFinalizeCheckpoint();
- fail("Expected IOException");
- }
- catch (IOException ignored) {
- // expected exception
- }
-
- verify(fs).delete(metadataPath, false);
- }
-
- /**
- * This test validates that a close operation can happen even while a 'closeAndGetHandle()'
- * call is in progress.
- *
- * <p>That behavior is essential for fast cancellation (concurrent cleanup).
- */
- @Test
- public void testCloseDoesNotLock() throws Exception {
- final Path checkpointDir = Path.fromLocalFile(tmp.newFolder());
- final Path metadataPath = new Path(checkpointDir, "this-is-ignored-anyways.file");
-
- final FileSystem fileSystem = spy(new TestFs((path) -> new BlockerStream()));
-
- final FixFileFsStateOutputStream checkpointStream =
- new FixFileFsStateOutputStream(fileSystem, metadataPath, checkpointDir);
-
- final OneShotLatch sync = new OneShotLatch();
-
- final CheckedThread thread = new CheckedThread() {
-
- @Override
- public void go() throws Exception {
- sync.trigger();
- // that call should now block, because it accesses the position
- checkpointStream.closeAndFinalizeCheckpoint();
- }
- };
- thread.start();
-
- sync.await();
- checkpointStream.close();
-
- // the thread may or may not fail, that depends on the thread race
- // it is not important for this test, important is that the thread does not freeze/lock up
- try {
- thread.sync();
- } catch (IOException ignored) {}
- }
-
- // ------------------------------------------------------------------------
- // utilities
- // ------------------------------------------------------------------------
-
- private static void readFully(InputStream in, byte[] buffer) throws IOException {
- int pos = 0;
- int remaining = buffer.length;
-
- while (remaining > 0) {
- int read = in.read(buffer, pos, remaining);
- if (read == -1) {
- throw new EOFException();
- }
-
- pos += read;
- remaining -= read;
- }
- }
-
- private static class BlockerStream extends FSDataOutputStream {
-
- private final OneShotLatch blocker = new OneShotLatch();
-
- @Override
- public long getPos() throws IOException {
- block();
- return 0L;
- }
-
- @Override
- public void write(int b) throws IOException {
- block();
- }
-
- @Override
- public void flush() throws IOException {
- block();
- }
-
- @Override
- public void sync() throws IOException {
- block();
- }
-
- @Override
- public void close() throws IOException {
- blocker.trigger();
- }
-
- private void block() throws IOException {
- try {
- blocker.await();
- } catch (InterruptedException e) {
- throw new IOException("interrupted");
- }
- 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/d85a62db/flink-runtime/src/test/java/org/apache/flink/runtime/state/filesystem/FsCheckpointMetadataOutputStreamTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/filesystem/FsCheckpointMetadataOutputStreamTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/filesystem/FsCheckpointMetadataOutputStreamTest.java
new file mode 100644
index 0000000..5a15b4b
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/filesystem/FsCheckpointMetadataOutputStreamTest.java
@@ -0,0 +1,308 @@
+/*
+ * 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.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.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 java.io.EOFException;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Random;
+
+import static org.junit.Assert.assertArrayEquals;
+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.Mockito.spy;
+import static org.mockito.Mockito.verify;
+
+/**
+ * Tests for the {@link FsCheckpointMetadataOutputStream}.
+ */
+public class FsCheckpointMetadataOutputStreamTest {
+
+ @Rule
+ public final TemporaryFolder tmp = new TemporaryFolder();
+
+ // ------------------------------------------------------------------------
+ // Tests
+ // ------------------------------------------------------------------------
+
+ /**
+ * Validates that even empty streams create a file and a file state handle.
+ */
+ @Test
+ public void testEmptyState() throws Exception {
+ final FileSystem fs = FileSystem.getLocalFileSystem();
+
+ final Path checkpointDir = Path.fromLocalFile(tmp.newFolder());
+ final Path metadataPath = new Path(checkpointDir, "myFileName");
+
+ final FsCompletedCheckpointStorageLocation location;
+ try (FsCheckpointMetadataOutputStream stream = new FsCheckpointMetadataOutputStream(fs, metadataPath, checkpointDir)) {
+ location = stream.closeAndFinalizeCheckpoint();
+ }
+
+ // must have created a handle
+ assertNotNull(location);
+ assertNotNull(location.getMetadataHandle());
+ assertEquals(metadataPath, location.getMetadataHandle().getFilePath());
+
+ // 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 = location.getMetadataHandle().openInputStream()) {
+ assertEquals(-1, in.read());
+ }
+ }
+
+ /**
+ * Simple write and read test.
+ */
+ @Test
+ public void testWriteAndRead() throws Exception {
+ final FileSystem fs = FileSystem.getLocalFileSystem();
+
+ 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 FsCompletedCheckpointStorageLocation completed;
+ try (FsCheckpointMetadataOutputStream stream = new FsCheckpointMetadataOutputStream(fs, metadataPath, checkpointDir)) {
+ for (int i = 0; i < data.length;) {
+ if (rnd.nextBoolean()) {
+ stream.write(data[i++]);
+ }
+ else {
+ int len = rnd.nextInt(Math.min(data.length - i, 32));
+ stream.write(data, i, len);
+ i += len;
+ }
+ }
+ completed = stream.closeAndFinalizeCheckpoint();
+ }
+
+ // (1) stream from handle must hold the contents
+ 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(completed.getMetadataHandle().getFilePath())) {
+ byte[] buffer = new byte[data.length];
+ readFully(in, buffer);
+ assertArrayEquals(data, buffer);
+ }
+ }
+
+ /**
+ * Tests that the underlying stream file is deleted upon calling close.
+ */
+ @Test
+ public void testCleanupWhenClosingStream() throws IOException {
+ final FileSystem fs = FileSystem.getLocalFileSystem();
+
+ final Path checkpointDir = Path.fromLocalFile(tmp.newFolder());
+ final Path metadataPath = new Path(checkpointDir, "nonCreativeTestFileName");
+
+ // write some test data and close the stream
+ try (FsCheckpointMetadataOutputStream stream = new FsCheckpointMetadataOutputStream(fs, metadataPath, checkpointDir)) {
+ Random rnd = new Random();
+ for (int i = 0; i < rnd.nextInt(1000); i++) {
+ stream.write(rnd.nextInt(100));
+ }
+ assertTrue(fs.exists(metadataPath));
+ }
+
+ assertFalse(fs.exists(metadataPath));
+ assertTrue(fs.exists(checkpointDir));
+ }
+
+ /**
+ * Tests that the underlying stream file is deleted if the closeAndGetHandle method fails.
+ */
+ @Test
+ public void testCleanupWhenFailingCloseAndGetHandle() throws IOException {
+ final Path checkpointDir = Path.fromLocalFile(tmp.newFolder());
+ final Path metadataPath = new Path(checkpointDir, "test_name");
+
+ final FileSystem fs = spy(new TestFs((path) -> new FailingCloseStream(new File(path.getPath()))));
+
+ FsCheckpointMetadataOutputStream stream = new FsCheckpointMetadataOutputStream(fs, metadataPath, checkpointDir);
+ stream.write(new byte[] {1, 2, 3, 4, 5});
+
+ try {
+ stream.closeAndFinalizeCheckpoint();
+ fail("Expected IOException");
+ }
+ catch (IOException ignored) {
+ // expected exception
+ }
+
+ verify(fs).delete(metadataPath, false);
+ }
+
+ /**
+ * This test validates that a close operation can happen even while a 'closeAndGetHandle()'
+ * call is in progress.
+ *
+ * <p>That behavior is essential for fast cancellation (concurrent cleanup).
+ */
+ @Test
+ public void testCloseDoesNotLock() throws Exception {
+ final Path checkpointDir = Path.fromLocalFile(tmp.newFolder());
+ final Path metadataPath = new Path(checkpointDir, "this-is-ignored-anyways.file");
+
+ final FileSystem fileSystem = spy(new TestFs((path) -> new BlockerStream()));
+
+ final FsCheckpointMetadataOutputStream checkpointStream =
+ new FsCheckpointMetadataOutputStream(fileSystem, metadataPath, checkpointDir);
+
+ final OneShotLatch sync = new OneShotLatch();
+
+ final CheckedThread thread = new CheckedThread() {
+
+ @Override
+ public void go() throws Exception {
+ sync.trigger();
+ // that call should now block, because it accesses the position
+ checkpointStream.closeAndFinalizeCheckpoint();
+ }
+ };
+ thread.start();
+
+ sync.await();
+ checkpointStream.close();
+
+ // the thread may or may not fail, that depends on the thread race
+ // it is not important for this test, important is that the thread does not freeze/lock up
+ try {
+ thread.sync();
+ } catch (IOException ignored) {}
+ }
+
+ // ------------------------------------------------------------------------
+ // utilities
+ // ------------------------------------------------------------------------
+
+ private static void readFully(InputStream in, byte[] buffer) throws IOException {
+ int pos = 0;
+ int remaining = buffer.length;
+
+ while (remaining > 0) {
+ int read = in.read(buffer, pos, remaining);
+ if (read == -1) {
+ throw new EOFException();
+ }
+
+ pos += read;
+ remaining -= read;
+ }
+ }
+
+ private static class BlockerStream extends FSDataOutputStream {
+
+ private final OneShotLatch blocker = new OneShotLatch();
+
+ @Override
+ public long getPos() throws IOException {
+ block();
+ return 0L;
+ }
+
+ @Override
+ public void write(int b) throws IOException {
+ block();
+ }
+
+ @Override
+ public void flush() throws IOException {
+ block();
+ }
+
+ @Override
+ public void sync() throws IOException {
+ block();
+ }
+
+ @Override
+ public void close() throws IOException {
+ blocker.trigger();
+ }
+
+ private void block() throws IOException {
+ try {
+ blocker.await();
+ } catch (InterruptedException e) {
+ throw new IOException("interrupted");
+ }
+ 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);
+ }
+ }
+}