From b32e376e9ef4a5cff8cd9670988db33c6c0fe7b9 Mon Sep 17 00:00:00 2001
From: Stephan Ewen <sewen@apache.org>
Date: Thu, 30 Oct 2014 20:50:15 +0100
Subject: [PATCH] Improve error messages in case of invalid file paths or URIs

---
 .../org/apache/flink/core/fs/FileSystem.java     | 13 ++++++++++---
 .../flink/core/fs/local/LocalFileSystem.java     |  4 ++--
 .../runtime/fs/hdfs/DistributedFileSystem.java   | 16 +++++++++++++++-
 3 files changed, 27 insertions(+), 6 deletions(-)

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 ef59996dd5c73..1b7b91e11cfc2 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
@@ -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:// or hdfs:// 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:///" + 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;
diff --git a/flink-core/src/main/java/org/apache/flink/core/fs/local/LocalFileSystem.java b/flink-core/src/main/java/org/apache/flink/core/fs/local/LocalFileSystem.java
index 7c547a2b9eb14..a33720b2805ea 100644
--- a/flink-core/src/main/java/org/apache/flink/core/fs/local/LocalFileSystem.java
+++ b/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.");
 		}
-
 	}
 
 
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/fs/hdfs/DistributedFileSystem.java b/flink-runtime/src/main/java/org/apache/flink/runtime/fs/hdfs/DistributedFileSystem.java
index d90da6e8bde18..d5f370f166b91 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/fs/hdfs/DistributedFileSystem.java
+++ b/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:///" + 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);
 			}
 		}