class FileSystem {

	public static FileSystem get(URI uri) throws IOException {
		FileSystem fs;

		URI asked = uri;
		synchronized (SYNCHRONIZATION_OBJECT) {

			if (uri.getScheme() == null) {
				try {
					if (defaultScheme == null) {
						defaultScheme = new URI(ConfigConstants.DEFAULT_FILESYSTEM_SCHEME);
					}

					uri = new URI(defaultScheme.getScheme(), null, defaultScheme.getHost(),
						defaultScheme.getPort(), uri.getPath(), null, null);

				} catch (URISyntaxException e) {
					try {
						if (defaultScheme.getScheme().equals("file")) {
							uri = new URI("file", null,
								new Path(new File(uri.getPath()).getAbsolutePath()).toUri().getPath(), null);
						}
					} catch (URISyntaxException ex) {
						// we tried to repair it, but could not. report the scheme error
						throw new IOException("The URI '" + uri.toString() + "' is not valid.");
					}
				}
			}

			if(uri.getScheme() == null) {
				throw new IOException("The URI '" + uri + "' is invalid.\n" +
					"The fs.default-scheme = " + defaultScheme + ", the requested URI = " + asked +
					", and the final URI = " + uri + ".");
			}

			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());

			// See if there is a file system object in the cache
			if (CACHE.containsKey(key)) {
				return CACHE.get(key);
			}

			// Try to create a new file system

			if (!FSDIRECTORY.containsKey(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) {
					// hadoop has support for the FileSystem
					FSKey wrappedKey = new FSKey(HADOOP_WRAPPER_SCHEME + "+" + uri.getScheme(), uri.getAuthority());
					if (CACHE.containsKey(wrappedKey)) {
						return CACHE.get(wrappedKey);
					}
					// cache didn't contain the file system. instantiate it:

					// by now we know that the HadoopFileSystem wrapper can wrap the file system.
					fs = instantiateHadoopFileSystemWrapper(wrapperClass);
					fs.initialize(uri);
					CACHE.put(wrappedKey, fs);

				} else {
					// we can not read from this file system.
					throw new IOException("No file system found with scheme " + uri.getScheme()
						+ ", referenced in file URI '" + uri.toString() + "'.");
				}
			} else {
				// we end up here if we have a file system with build-in flink support.
				String fsClass = FSDIRECTORY.get(uri.getScheme());
				if (fsClass.equals(HADOOP_WRAPPER_FILESYSTEM_CLASS)) {
					fs = instantiateHadoopFileSystemWrapper(null);
				} else {
					fs = instantiateFileSystem(fsClass);
				}
				// Initialize new file system object
				fs.initialize(uri);

				// Add new file system object to cache
				CACHE.put(key, fs);
			}
		}

		return fs;
	}

}