You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by kk...@apache.org on 2018/12/05 14:11:40 UTC
[flink] 01/02: [FLINK-10522][fs-connector] Check if
RecoverableWriter supportsResume() and act accordingly.
This is an automated email from the ASF dual-hosted git repository.
kkloudas pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 600bd9cdbb42b709cfa6fa313648fcea317ad781
Author: Kostas Kloudas <kk...@gmail.com>
AuthorDate: Tue Dec 4 17:35:42 2018 +0100
[FLINK-10522][fs-connector] Check if RecoverableWriter supportsResume() and act accordingly.
This closes #7047.
---
.../api/functions/sink/filesystem/Bucket.java | 14 +-
.../api/functions/sink/filesystem/BucketTest.java | 185 ++++++++++++++++++++-
.../sink/filesystem/utils/NoOpCommitter.java | 49 ++++++
.../sink/filesystem/utils/NoOpRecoverable.java | 32 ++++
.../utils/NoOpRecoverableFsDataOutputStream.java | 67 ++++++++
.../filesystem/utils/NoOpRecoverableWriter.java | 75 +++++++++
6 files changed, 418 insertions(+), 4 deletions(-)
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/Bucket.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/Bucket.java
index b59c84e..3252d9c 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/Bucket.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/Bucket.java
@@ -135,9 +135,17 @@ public class Bucket<IN, BucketID> {
// we try to resume the previous in-progress file
final ResumeRecoverable resumable = state.getInProgressResumableFile();
- final RecoverableFsDataOutputStream stream = fsWriter.recover(resumable);
- inProgressPart = partFileFactory.resumeFrom(
- bucketId, stream, resumable, state.getInProgressFileCreationTime());
+
+ if (fsWriter.supportsResume()) {
+ final RecoverableFsDataOutputStream stream = fsWriter.recover(resumable);
+ inProgressPart = partFileFactory.resumeFrom(
+ bucketId, stream, resumable, state.getInProgressFileCreationTime());
+ } else {
+ // if the writer does not support resume, then we close the
+ // in-progress part and commit it, as done in the case of pending files.
+
+ fsWriter.recoverForCommit(resumable).commitAfterRecovery();
+ }
if (fsWriter.requiresCleanupOfRecoverableState()) {
fsWriter.cleanupRecoverableState(resumable);
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/BucketTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/BucketTest.java
index f328fd7..308bc31 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/BucketTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/BucketTest.java
@@ -21,25 +21,36 @@ package org.apache.flink.streaming.api.functions.sink.filesystem;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.Path;
+import org.apache.flink.core.fs.RecoverableFsDataOutputStream;
import org.apache.flink.core.fs.RecoverableWriter;
import org.apache.flink.core.fs.local.LocalFileSystem;
import org.apache.flink.core.fs.local.LocalRecoverableWriter;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;
+import org.apache.flink.streaming.api.functions.sink.filesystem.utils.NoOpCommitter;
+import org.apache.flink.streaming.api.functions.sink.filesystem.utils.NoOpRecoverable;
+import org.apache.flink.streaming.api.functions.sink.filesystem.utils.NoOpRecoverableFsDataOutputStream;
+import org.apache.flink.streaming.api.functions.sink.filesystem.utils.NoOpRecoverableWriter;
import org.hamcrest.Description;
import org.hamcrest.TypeSafeMatcher;
+import org.junit.Assert;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import java.io.File;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import static org.apache.flink.util.Preconditions.checkArgument;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.fail;
/**
- * Tests for the {@link Bucket}.
+ * Tests for the {@code Bucket}.
*/
public class BucketTest {
@@ -128,6 +139,37 @@ public class BucketTest {
assertThat(recoverableWriter, hasCalledDiscard(0)); // we have no in-progress file.
}
+ // --------------------------- Checking Restore ---------------------------
+
+ @Test
+ public void inProgressFileShouldBeCommittedIfWriterDoesNotSupportResume() throws IOException {
+ final StubNonResumableWriter nonResumableWriter = new StubNonResumableWriter();
+ final Bucket<String, String> bucket = getRestoredBucketWithOnlyInProgressPart(nonResumableWriter);
+
+ Assert.assertThat(nonResumableWriter, hasMethodCallCountersEqualTo(1, 0, 1));
+ Assert.assertThat(bucket, hasNullInProgressFile(true));
+ }
+
+ @Test
+ public void inProgressFileShouldBeRestoredIfWriterSupportsResume() throws IOException {
+ final StubResumableWriter resumableWriter = new StubResumableWriter();
+ final Bucket<String, String> bucket = getRestoredBucketWithOnlyInProgressPart(resumableWriter);
+
+ Assert.assertThat(resumableWriter, hasMethodCallCountersEqualTo(1, 1, 0));
+ Assert.assertThat(bucket, hasNullInProgressFile(false));
+ }
+
+ @Test
+ public void pendingFilesShouldBeRestored() throws IOException {
+ final int expectedRecoverForCommitCounter = 10;
+
+ final StubNonResumableWriter writer = new StubNonResumableWriter();
+ final Bucket<String, String> bucket = getRestoredBucketWithOnlyPendingParts(writer, expectedRecoverForCommitCounter);
+
+ Assert.assertThat(writer, hasMethodCallCountersEqualTo(0, 0, expectedRecoverForCommitCounter));
+ Assert.assertThat(bucket, hasNullInProgressFile(true));
+ }
+
// ------------------------------- Matchers --------------------------------
private static TypeSafeMatcher<TestRecoverableWriter> hasCalledDiscard(int times) {
@@ -175,6 +217,47 @@ public class BucketTest {
};
}
+ private static TypeSafeMatcher<Bucket<String, String>> hasNullInProgressFile(final boolean isNull) {
+
+ return new TypeSafeMatcher<Bucket<String, String>>() {
+ @Override
+ protected boolean matchesSafely(Bucket<String, String> bucket) {
+ final PartFileWriter<String, String> inProgressPart = bucket.getInProgressPart();
+ return isNull == (inProgressPart == null);
+ }
+
+ @Override
+ public void describeTo(Description description) {
+ description.appendText("a Bucket with its inProgressPart being ")
+ .appendText(isNull ? " null." : " not null.");
+ }
+ };
+ }
+
+ private static TypeSafeMatcher<BaseStubWriter> hasMethodCallCountersEqualTo(
+ final int supportsResumeCalls,
+ final int recoverCalls,
+ final int recoverForCommitCalls) {
+
+ return new TypeSafeMatcher<BaseStubWriter>() {
+ @Override
+ protected boolean matchesSafely(BaseStubWriter writer) {
+ return writer.getSupportsResumeCallCounter() == supportsResumeCalls &&
+ writer.getRecoverCallCounter() == recoverCalls &&
+ writer.getRecoverForCommitCallCounter() == recoverForCommitCalls;
+ }
+
+ @Override
+ public void describeTo(Description description) {
+ description.appendText("a Writer where:")
+ .appendText(" supportsResume was called ").appendValue(supportsResumeCalls).appendText(" times,")
+ .appendText(" recover was called ").appendValue(recoverCalls).appendText(" times,")
+ .appendText(" and recoverForCommit was called ").appendValue(recoverForCommitCalls).appendText(" times.")
+ .appendText("'");
+ }
+ };
+ }
+
// ------------------------------- Mock Classes --------------------------------
private static class TestRecoverableWriter extends LocalRecoverableWriter {
@@ -207,6 +290,81 @@ public class BucketTest {
}
}
+ /**
+ * A test implementation of a {@link RecoverableWriter} that does not support
+ * resuming, i.e. keep on writing to the in-progress file at the point we were
+ * before the failure.
+ */
+ private static class StubResumableWriter extends BaseStubWriter {
+
+ StubResumableWriter() {
+ super(true);
+ }
+ }
+
+ /**
+ * A test implementation of a {@link RecoverableWriter} that does not support
+ * resuming, i.e. keep on writing to the in-progress file at the point we were
+ * before the failure.
+ */
+ private static class StubNonResumableWriter extends BaseStubWriter {
+
+ StubNonResumableWriter() {
+ super(false);
+ }
+ }
+
+ /**
+ * A test implementation of a {@link RecoverableWriter} that does not support
+ * resuming, i.e. keep on writing to the in-progress file at the point we were
+ * before the failure.
+ */
+ private static class BaseStubWriter extends NoOpRecoverableWriter {
+
+ private final boolean supportsResume;
+
+ private int supportsResumeCallCounter = 0;
+
+ private int recoverCallCounter = 0;
+
+ private int recoverForCommitCallCounter = 0;
+
+ private BaseStubWriter(final boolean supportsResume) {
+ this.supportsResume = supportsResume;
+ }
+
+ int getSupportsResumeCallCounter() {
+ return supportsResumeCallCounter;
+ }
+
+ int getRecoverCallCounter() {
+ return recoverCallCounter;
+ }
+
+ int getRecoverForCommitCallCounter() {
+ return recoverForCommitCallCounter;
+ }
+
+ @Override
+ public RecoverableFsDataOutputStream recover(ResumeRecoverable resumable) throws IOException {
+ recoverCallCounter++;
+ return new NoOpRecoverableFsDataOutputStream();
+ }
+
+ @Override
+ public RecoverableFsDataOutputStream.Committer recoverForCommit(CommitRecoverable resumable) throws IOException {
+ checkArgument(resumable instanceof NoOpRecoverable);
+ recoverForCommitCallCounter++;
+ return new NoOpCommitter();
+ }
+
+ @Override
+ public boolean supportsResume() {
+ supportsResumeCallCounter++;
+ return supportsResume;
+ }
+ }
+
// ------------------------------- Utility Methods --------------------------------
private static final String bucketId = "testing-bucket";
@@ -260,4 +418,29 @@ public class BucketTest {
}
return null;
}
+
+ private Bucket<String, String> getRestoredBucketWithOnlyInProgressPart(final BaseStubWriter writer) throws IOException {
+ final BucketState<String> stateWithOnlyInProgressFile =
+ new BucketState<>("test", new Path(), 12345L, new NoOpRecoverable(), new HashMap<>());
+ return Bucket.restore(writer, 0, 1L, partFileFactory, rollingPolicy, stateWithOnlyInProgressFile);
+ }
+
+ private Bucket<String, String> getRestoredBucketWithOnlyPendingParts(final BaseStubWriter writer, final int numberOfPendingParts) throws IOException {
+ final Map<Long, List<RecoverableWriter.CommitRecoverable>> completePartsPerCheckpoint =
+ createPendingPartsPerCheckpoint(numberOfPendingParts);
+
+ final BucketState<String> initStateWithOnlyInProgressFile =
+ new BucketState<>("test", new Path(), 12345L, null, completePartsPerCheckpoint);
+ return Bucket.restore(writer, 0, 1L, partFileFactory, rollingPolicy, initStateWithOnlyInProgressFile);
+ }
+
+ private Map<Long, List<RecoverableWriter.CommitRecoverable>> createPendingPartsPerCheckpoint(int noOfCheckpoints) {
+ final Map<Long, List<RecoverableWriter.CommitRecoverable>> pendingCommittablesPerCheckpoint = new HashMap<>();
+ for (int checkpointId = 0; checkpointId < noOfCheckpoints; checkpointId++) {
+ final List<RecoverableWriter.CommitRecoverable> pending = new ArrayList<>();
+ pending.add(new NoOpRecoverable());
+ pendingCommittablesPerCheckpoint.put((long) checkpointId, pending);
+ }
+ return pendingCommittablesPerCheckpoint;
+ }
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/utils/NoOpCommitter.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/utils/NoOpCommitter.java
new file mode 100644
index 0000000..06005a1
--- /dev/null
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/utils/NoOpCommitter.java
@@ -0,0 +1,49 @@
+/*
+ * 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.streaming.api.functions.sink.filesystem.utils;
+
+import org.apache.flink.core.fs.RecoverableFsDataOutputStream;
+import org.apache.flink.core.fs.RecoverableWriter;
+
+import java.io.IOException;
+
+/**
+ * An implementation of the {@link RecoverableFsDataOutputStream.Committer committer}
+ * that does nothing.
+ *
+ * <p>This is to avoid to have to implement all methods for every implementation
+ * used in tests.
+ */
+public class NoOpCommitter implements RecoverableFsDataOutputStream.Committer {
+
+ @Override
+ public void commit() throws IOException {
+
+ }
+
+ @Override
+ public void commitAfterRecovery() throws IOException {
+
+ }
+
+ @Override
+ public RecoverableWriter.CommitRecoverable getRecoverable() {
+ return null;
+ }
+}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/utils/NoOpRecoverable.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/utils/NoOpRecoverable.java
new file mode 100644
index 0000000..e00d4c4
--- /dev/null
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/utils/NoOpRecoverable.java
@@ -0,0 +1,32 @@
+/*
+ * 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.streaming.api.functions.sink.filesystem.utils;
+
+import org.apache.flink.core.fs.RecoverableWriter;
+
+/**
+ * An implementation of the {@link RecoverableWriter.ResumeRecoverable ResumeRecoverable}
+ * that does nothing.
+ *
+ * <p>This is to avoid to have to implement all methods for every implementation
+ * used in tests.
+ */
+public class NoOpRecoverable implements RecoverableWriter.ResumeRecoverable {
+
+}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/utils/NoOpRecoverableFsDataOutputStream.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/utils/NoOpRecoverableFsDataOutputStream.java
new file mode 100644
index 0000000..a549896
--- /dev/null
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/utils/NoOpRecoverableFsDataOutputStream.java
@@ -0,0 +1,67 @@
+/*
+ * 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.streaming.api.functions.sink.filesystem.utils;
+
+import org.apache.flink.core.fs.RecoverableFsDataOutputStream;
+import org.apache.flink.core.fs.RecoverableWriter;
+
+import java.io.IOException;
+
+/**
+ * A default implementation of the {@link RecoverableFsDataOutputStream} that does nothing.
+ *
+ * <p>This is to avoid to have to implement all methods for every implementation
+ * used in tests.
+ */
+public class NoOpRecoverableFsDataOutputStream extends RecoverableFsDataOutputStream {
+ @Override
+ public RecoverableWriter.ResumeRecoverable persist() throws IOException {
+ return null;
+ }
+
+ @Override
+ public Committer closeForCommit() throws IOException {
+ return null;
+ }
+
+ @Override
+ public void close() throws IOException {
+
+ }
+
+ @Override
+ public long getPos() throws IOException {
+ return 0;
+ }
+
+ @Override
+ public void flush() throws IOException {
+
+ }
+
+ @Override
+ public void sync() throws IOException {
+
+ }
+
+ @Override
+ public void write(int b) throws IOException {
+
+ }
+}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/utils/NoOpRecoverableWriter.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/utils/NoOpRecoverableWriter.java
new file mode 100644
index 0000000..e21da2a
--- /dev/null
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/utils/NoOpRecoverableWriter.java
@@ -0,0 +1,75 @@
+/*
+ * 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.streaming.api.functions.sink.filesystem.utils;
+
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.core.fs.RecoverableFsDataOutputStream;
+import org.apache.flink.core.fs.RecoverableWriter;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+
+import java.io.IOException;
+
+/**
+ * A default implementation of the {@link RecoverableWriter} that does nothing.
+ *
+ * <p>This is to avoid to have to implement all methods for every implementation
+ * used in tests.
+ */
+public class NoOpRecoverableWriter implements RecoverableWriter {
+
+ @Override
+ public RecoverableFsDataOutputStream open(Path path) throws IOException {
+ return null;
+ }
+
+ @Override
+ public RecoverableFsDataOutputStream recover(RecoverableWriter.ResumeRecoverable resumable) throws IOException {
+ return null;
+ }
+
+ @Override
+ public boolean requiresCleanupOfRecoverableState() {
+ return false;
+ }
+
+ @Override
+ public boolean cleanupRecoverableState(ResumeRecoverable resumable) throws IOException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public RecoverableFsDataOutputStream.Committer recoverForCommit(RecoverableWriter.CommitRecoverable resumable) throws IOException {
+ return null;
+ }
+
+ @Override
+ public SimpleVersionedSerializer<RecoverableWriter.CommitRecoverable> getCommitRecoverableSerializer() {
+ return null;
+ }
+
+ @Override
+ public SimpleVersionedSerializer<RecoverableWriter.ResumeRecoverable> getResumeRecoverableSerializer() {
+ return null;
+ }
+
+ @Override
+ public boolean supportsResume() {
+ return false;
+ }
+}