You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by sr...@apache.org on 2018/03/05 11:18:05 UTC
flink git commit: [FLINK-8818][yarn/s3][tests] harden
YarnFileStageTest upload test for eventual consistent read-after-write
Repository: flink
Updated Branches:
refs/heads/master 56fef58a3 -> c74d8cac2
[FLINK-8818][yarn/s3][tests] harden YarnFileStageTest upload test for eventual consistent read-after-write
In case the newly written object cannot be read (yet), we do 4 more retries to
retrieve the value and wait 50ms each. While this does not solve all the cases
it should make the (rare) case of the written object not being available for
read even more unlikely.
This closes #5601.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/c74d8cac
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/c74d8cac
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/c74d8cac
Branch: refs/heads/master
Commit: c74d8cac25e8c025ced11a9c03cd0cf07a8c2d6b
Parents: 56fef58
Author: Nico Kruber <ni...@data-artisans.com>
Authored: Tue Feb 27 17:29:00 2018 +0100
Committer: Stefan Richter <s....@data-artisans.com>
Committed: Mon Mar 5 12:17:53 2018 +0100
----------------------------------------------------------------------
.../apache/flink/yarn/YarnFileStageTest.java | 25 ++++++++++++++------
1 file changed, 18 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/c74d8cac/flink-yarn/src/test/java/org/apache/flink/yarn/YarnFileStageTest.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/test/java/org/apache/flink/yarn/YarnFileStageTest.java b/flink-yarn/src/test/java/org/apache/flink/yarn/YarnFileStageTest.java
index 5cbe1be..527782c 100644
--- a/flink-yarn/src/test/java/org/apache/flink/yarn/YarnFileStageTest.java
+++ b/flink-yarn/src/test/java/org/apache/flink/yarn/YarnFileStageTest.java
@@ -41,6 +41,7 @@ import org.junit.rules.TemporaryFolder;
import java.io.DataOutputStream;
import java.io.File;
+import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.util.ArrayList;
import java.util.Collections;
@@ -200,13 +201,23 @@ public class YarnFileStageTest extends TestLogger {
while (targetFilesIterator.hasNext()) {
LocatedFileStatus targetFile = targetFilesIterator.next();
- try (FSDataInputStream in = targetFileSystem.open(targetFile.getPath())) {
- String absolutePathString = targetFile.getPath().toString();
- String relativePath = absolutePathString.substring(workDirPrefixLength);
- targetFiles.put(relativePath, in.readUTF());
-
- assertEquals("extraneous data in file " + relativePath, -1, in.read());
- }
+ int retries = 5;
+ do {
+ try (FSDataInputStream in = targetFileSystem.open(targetFile.getPath())) {
+ String absolutePathString = targetFile.getPath().toString();
+ String relativePath = absolutePathString.substring(workDirPrefixLength);
+ targetFiles.put(relativePath, in.readUTF());
+
+ assertEquals("extraneous data in file " + relativePath, -1, in.read());
+ break;
+ } catch (FileNotFoundException e) {
+ // For S3, read-after-write may be eventually consistent, i.e. when trying
+ // to access the object before writing it; see
+ // https://docs.aws.amazon.com/AmazonS3/latest/dev/Introduction.html#ConsistencyModel
+ // -> try again a bit later
+ Thread.sleep(50);
+ }
+ } while ((retries--) > 0);
}
assertThat(targetFiles, equalTo(srcFiles));