You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by al...@apache.org on 2016/07/05 11:50:35 UTC
[1/3] flink git commit: [FLINK-4115] Remove filesystem initialisation
from FsStateBackend constructor
Repository: flink
Updated Branches:
refs/heads/master 647b4c577 -> 5c0c7e36e
[FLINK-4115] Remove filesystem initialisation from FsStateBackend constructor
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/989c67a1
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/989c67a1
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/989c67a1
Branch: refs/heads/master
Commit: 989c67a13b6c23b246cb3444ca6c9406f6e4be79
Parents: a3d1a17
Author: Josh <jo...@permutive.com>
Authored: Thu Jun 23 22:22:40 2016 +0100
Committer: Aljoscha Krettek <al...@gmail.com>
Committed: Tue Jul 5 10:39:57 2016 +0200
----------------------------------------------------------------------
.../org/apache/flink/runtime/state/filesystem/FsStateBackend.java | 1 -
1 file changed, 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/989c67a1/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsStateBackend.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsStateBackend.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsStateBackend.java
index 1f198ec..f4de48c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsStateBackend.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsStateBackend.java
@@ -173,7 +173,6 @@ public class FsStateBackend extends AbstractStateBackend {
this.fileStateThreshold = fileStateSizeThreshold;
this.basePath = validateAndNormalizeUri(checkpointDataUri);
- this.filesystem = this.basePath.getFileSystem();
}
/**
[2/3] flink git commit: [FLINK-4115] Skip filesystem checks for
filesystems with no built-in support
Posted by al...@apache.org.
[FLINK-4115] Skip filesystem checks for filesystems with no built-in support
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/a3d1a174
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/a3d1a174
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/a3d1a174
Branch: refs/heads/master
Commit: a3d1a17406749e41185a77a64c23fc1cb3eabc76
Parents: 647b4c5
Author: Josh <jo...@permutive.com>
Authored: Thu Jun 23 22:18:02 2016 +0100
Committer: Aljoscha Krettek <al...@gmail.com>
Committed: Tue Jul 5 10:39:57 2016 +0200
----------------------------------------------------------------------
.../org/apache/flink/core/fs/FileSystem.java | 13 ++++++-
.../state/filesystem/FsStateBackend.java | 41 +++++++++++---------
2 files changed, 35 insertions(+), 19 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/a3d1a174/flink-core/src/main/java/org/apache/flink/core/fs/FileSystem.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/core/fs/FileSystem.java b/flink-core/src/main/java/org/apache/flink/core/fs/FileSystem.java
index 3e36b88..1844d64 100644
--- a/flink-core/src/main/java/org/apache/flink/core/fs/FileSystem.java
+++ b/flink-core/src/main/java/org/apache/flink/core/fs/FileSystem.java
@@ -275,7 +275,7 @@ public abstract class FileSystem {
// Try to create a new file system
- if (!FSDIRECTORY.containsKey(uri.getScheme())) {
+ if (!isFlinkSupportedScheme(uri.getScheme())) {
// no build in support for this file system. Falling back to Hadoop's FileSystem impl.
Class<?> wrapperClass = getHadoopWrapperClassNameForFileSystem(uri.getScheme());
if (wrapperClass != null) {
@@ -315,6 +315,17 @@ public abstract class FileSystem {
return fs;
}
+ /**
+ * Returns a boolean indicating whether a scheme has built-in Flink support.
+ *
+ * @param scheme
+ * a file system scheme
+ * @return a boolean indicating whether the provided scheme has built-in Flink support
+ */
+ public static boolean isFlinkSupportedScheme(String scheme) {
+ return FSDIRECTORY.containsKey(scheme);
+ }
+
//Class must implement Hadoop FileSystem interface. The class is not avaiable in 'flink-core'.
private static FileSystem instantiateHadoopFileSystemWrapper(Class<?> wrappedFileSystem) throws IOException {
try {
http://git-wip-us.apache.org/repos/asf/flink/blob/a3d1a174/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsStateBackend.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsStateBackend.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsStateBackend.java
index 77d540b..1f198ec 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsStateBackend.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsStateBackend.java
@@ -374,26 +374,31 @@ public class FsStateBackend extends AbstractStateBackend {
throw new IllegalArgumentException("Cannot use the root directory for checkpoints.");
}
- // we do a bit of work to make sure that the URI for the filesystem refers to exactly the same
- // (distributed) filesystem on all hosts and includes full host/port information, even if the
- // original URI did not include that. We count on the filesystem loading from the configuration
- // to fill in the missing data.
-
- // try to grab the file system for this path/URI
- FileSystem filesystem = FileSystem.get(checkpointDataUri);
- if (filesystem == null) {
- throw new IOException("Could not find a file system for the given scheme in the available configurations.");
- }
+ if (!FileSystem.isFlinkSupportedScheme(checkpointDataUri.getScheme())) {
+ // skip verification checks for non-flink supported filesystem
+ // this is because the required filesystem classes may not be available to the flink client
+ return new Path(checkpointDataUri);
+ } else {
+ // we do a bit of work to make sure that the URI for the filesystem refers to exactly the same
+ // (distributed) filesystem on all hosts and includes full host/port information, even if the
+ // original URI did not include that. We count on the filesystem loading from the configuration
+ // to fill in the missing data.
+
+ // try to grab the file system for this path/URI
+ FileSystem filesystem = FileSystem.get(checkpointDataUri);
+ if (filesystem == null) {
+ throw new IOException("Could not find a file system for the given scheme in the available configurations.");
+ }
- URI fsURI = filesystem.getUri();
- try {
- URI baseURI = new URI(fsURI.getScheme(), fsURI.getAuthority(), path, null, null);
- return new Path(baseURI);
- }
- catch (URISyntaxException e) {
- throw new IOException(
+ URI fsURI = filesystem.getUri();
+ try {
+ URI baseURI = new URI(fsURI.getScheme(), fsURI.getAuthority(), path, null, null);
+ return new Path(baseURI);
+ } catch (URISyntaxException e) {
+ throw new IOException(
String.format("Cannot create file system URI for checkpointDataUri %s and filesystem URI %s",
- checkpointDataUri, fsURI), e);
+ checkpointDataUri, fsURI), e);
+ }
}
}
[3/3] flink git commit: Change Validation Exception to LOG.warn() in
FsStateBackend
Posted by al...@apache.org.
Change Validation Exception to LOG.warn() in FsStateBackend
Before, job would fail with an exception if the checkpoint filesystem
was not accessible from the client. This is preventing some correct jobs
from running.
Now, only print a warning.
This closes #2157
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/5c0c7e36
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/5c0c7e36
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/5c0c7e36
Branch: refs/heads/master
Commit: 5c0c7e36e8984dfd29224389212d2d04e24b4c8b
Parents: 989c67a
Author: Aljoscha Krettek <al...@gmail.com>
Authored: Tue Jul 5 10:37:08 2016 +0200
Committer: Aljoscha Krettek <al...@gmail.com>
Committed: Tue Jul 5 13:46:22 2016 +0200
----------------------------------------------------------------------
.../runtime/state/filesystem/FsStateBackend.java | 18 ++++++++++++++----
1 file changed, 14 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/5c0c7e36/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsStateBackend.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsStateBackend.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsStateBackend.java
index f4de48c..8a8a26d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsStateBackend.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsStateBackend.java
@@ -386,7 +386,12 @@ public class FsStateBackend extends AbstractStateBackend {
// try to grab the file system for this path/URI
FileSystem filesystem = FileSystem.get(checkpointDataUri);
if (filesystem == null) {
- throw new IOException("Could not find a file system for the given scheme in the available configurations.");
+ String reason = "Could not find a file system for the given scheme in" +
+ "the available configurations.";
+ LOG.warn("Could not verify checkpoint path. This might be caused by a genuine " +
+ "problem or by the fact that the file system is not accessible from the " +
+ "client. Reason:{}", reason);
+ return new Path(checkpointDataUri);
}
URI fsURI = filesystem.getUri();
@@ -394,9 +399,14 @@ public class FsStateBackend extends AbstractStateBackend {
URI baseURI = new URI(fsURI.getScheme(), fsURI.getAuthority(), path, null, null);
return new Path(baseURI);
} catch (URISyntaxException e) {
- throw new IOException(
- String.format("Cannot create file system URI for checkpointDataUri %s and filesystem URI %s",
- checkpointDataUri, fsURI), e);
+ String reason = String.format(
+ "Cannot create file system URI for checkpointDataUri %s and filesystem URI %s: " + e.toString(),
+ checkpointDataUri,
+ fsURI);
+ LOG.warn("Could not verify checkpoint path. This might be caused by a genuine " +
+ "problem or by the fact that the file system is not accessible from the " +
+ "client. Reason: {}", reason);
+ return new Path(checkpointDataUri);
}
}
}