Improve error messages in case of invalid file paths or URIs

This closes apache#170

flink-core/src/main/java/org/apache/flink/core/fs/FileSystem.java

@@ -209,10 +209,17 @@ public static FileSystem get(URI uri) throws IOException {
  }
  catch (URISyntaxException e) {
  // we tried to repair it, but could not. report the scheme error
- throw new IOException("FileSystem: Scheme is null. file:https:// or hdfs:https:// are example schemes. "
- + "Failed for " + uri.toString() + ".");
+ throw new IOException("The file URI '" + uri.toString() + "' is not valid. "
+ + " File URIs need to specify aboslute file paths.");
  }
  }
+
+ if (uri.getScheme().equals("file") && uri.getAuthority() != null && !uri.getAuthority().isEmpty()) {
+ String supposedUri = "file:https:///" + uri.getAuthority() + uri.getPath();
+
+ throw new IOException("Found local file path with authority '" + uri.getAuthority() + "' in path '"
+ + uri.toString() + "'. Hint: Did you forget a slash? (correct path would be '" + supposedUri + "')");
+ }
 
  final FSKey key = new FSKey(uri.getScheme(), uri.getAuthority());
 
@@ -224,7 +231,7 @@ public static FileSystem get(URI uri) throws IOException {
  // Try to create a new file system
  if (!FSDIRECTORY.containsKey(uri.getScheme())) {
  throw new IOException("No file system found with scheme " + uri.getScheme()
- + ". Failed for " + uri.toString() + ".");
+  + ", referenced in file URI '" + uri.toString() + "'.");
  }
 
  Class<? extends FileSystem> fsClass = null;

flink-core/src/main/java/org/apache/flink/core/fs/local/LocalFileSystem.java

@@ -101,11 +101,11 @@ public FileStatus getFileStatus(Path f) throws IOException {
  final File path = pathToFile(f);
  if (path.exists()) {
  return new LocalFileStatus(pathToFile(f), this);
- } else {
+ }
+ else {
  throw new FileNotFoundException("File " + f + " does not exist or the user running "
  + "Flink ('"+System.getProperty("user.name")+"') has insufficient permissions to access it.");
  }
-
  }
 
 

flink-runtime/src/main/java/org/apache/flink/runtime/fs/hdfs/DistributedFileSystem.java

@@ -23,6 +23,7 @@
 import java.io.IOException;
 import java.lang.reflect.Method;
 import java.net.URI;
+import java.net.UnknownHostException;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -255,6 +256,7 @@ public URI getUri() {
 
  @Override
  public void initialize(URI path) throws IOException {
+
  // For HDFS we have to have an authority
  if (path.getAuthority() == null) {
 
@@ -301,8 +303,20 @@ public void initialize(URI path) throws IOException {
  try {
  this.fs.initialize(path, this.conf);
  }
+ catch (UnknownHostException e) {
+ String message = "The HDFS namenode host at '" + path.getAuthority()
+ + "', specified by file path '" + path.toString() + "', cannot be resolved"
+ + (e.getMessage() != null ? ": " + e.getMessage() : ".");
+
+ if (path.getPort() == -1) {
+ message += " Hint: Have you forgotten a slash? (correct URI would be 'hdfs:https:///" + path.getAuthority() + path.getPath() + "' ?)";
+ }
+
+ throw new IOException(message, e);
+ }
  catch (Exception e) {
- throw new IOException("The given file URI (" + path.toString() + ") described the host and port of an HDFS Namenode, but the File System could not be initialized with that address"
+ throw new IOException("The given file URI (" + path.toString() + ") points to the HDFS Namenode at "
+ + path.getAuthority() + ", but the File System could not be initialized with that address"
  + (e.getMessage() != null ? ": " + e.getMessage() : "."), e);
  }
  }