You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2017/10/06 10:05:50 UTC
[1/4] flink git commit: [FLINK-7643] [core] Drop eager checks for
file system support.
Repository: flink
Updated Branches:
refs/heads/master 84a07a34a -> 536675b03
[FLINK-7643] [core] Drop eager checks for file system support.
Some places validate if the file URIs are resolvable on the client. This leads to
problems when file systems are not accessible from the client, when the full libraries for
the file systems are not present on the client (for example often the case in cloud setups),
or when the configuration on the client is different from the nodes/containers that will
execute the application.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/a5ef09bb
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/a5ef09bb
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/a5ef09bb
Branch: refs/heads/master
Commit: a5ef09bb601cdd77fcb94e9ce633fdf979031aaf
Parents: 3b78684
Author: Stephan Ewen <se...@apache.org>
Authored: Mon Oct 2 16:30:07 2017 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Thu Oct 5 19:14:45 2017 +0200
----------------------------------------------------------------------
.../org/apache/flink/core/fs/FileSystem.java | 11 -----
.../state/filesystem/FsStateBackend.java | 44 +-------------------
.../runtime/jobmanager/MemoryArchivist.scala | 11 ++---
3 files changed, 8 insertions(+), 58 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/a5ef09bb/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 7c69425..672ebbb 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
@@ -406,17 +406,6 @@ public abstract class FileSystem {
return FileSystemSafetyNet.wrapWithSafetyNetWhenActivated(getUnguardedFileSystem(uri));
}
- /**
- * 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/a5ef09bb/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 e320bf3..ddfa85c 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
@@ -31,12 +31,9 @@ import org.apache.flink.runtime.state.DefaultOperatorStateBackend;
import org.apache.flink.runtime.state.KeyGroupRange;
import org.apache.flink.runtime.state.OperatorStateBackend;
import org.apache.flink.runtime.state.heap.HeapKeyedStateBackend;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.net.URI;
-import java.net.URISyntaxException;
import static org.apache.flink.util.Preconditions.checkArgument;
@@ -53,8 +50,6 @@ public class FsStateBackend extends AbstractStateBackend {
private static final long serialVersionUID = -8191916350224044011L;
- private static final Logger LOG = LoggerFactory.getLogger(FsStateBackend.class);
-
/** By default, state smaller than 1024 bytes will not be written to files, but
* will be stored directly with the metadata */
public static final int DEFAULT_FILE_STATE_THRESHOLD = 1024;
@@ -341,7 +336,7 @@ public class FsStateBackend extends AbstractStateBackend {
* @throws IllegalArgumentException Thrown, if the URI misses scheme or path.
* @throws IOException Thrown, if no file system can be found for the URI's scheme.
*/
- public static Path validateAndNormalizeUri(URI checkpointDataUri) throws IOException {
+ private static Path validateAndNormalizeUri(URI checkpointDataUri) throws IOException {
final String scheme = checkpointDataUri.getScheme();
final String path = checkpointDataUri.getPath();
@@ -358,41 +353,6 @@ public class FsStateBackend extends AbstractStateBackend {
throw new IllegalArgumentException("Cannot use the root directory for checkpoints.");
}
- 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) {
- 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();
- try {
- URI baseURI = new URI(fsURI.getScheme(), fsURI.getAuthority(), path, null, null);
- return new Path(baseURI);
- } catch (URISyntaxException 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);
- }
- }
+ return new Path(checkpointDataUri);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/a5ef09bb/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/MemoryArchivist.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/MemoryArchivist.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/MemoryArchivist.scala
index c963238..7a10d01 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/MemoryArchivist.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/MemoryArchivist.scala
@@ -291,11 +291,12 @@ class MemoryArchivist(
throw new IllegalArgumentException("Cannot use the root directory for storing job archives.")
}
- if (!FileSystem.isFlinkSupportedScheme(archivePathUri.getScheme)) {
- // skip verification checks for non-flink supported filesystem
- // this is because the required filesystem classes may not be available to the flink client
- throw new IllegalArgumentException("No file system found with scheme " + scheme
- + ", referenced in file URI '" + archivePathUri.toString + "'.")
+ try {
+ FileSystem.get(archivePathUri)
+ }
+ catch {
+ case e: Exception =>
+ throw new IllegalArgumentException(s"No file system found for URI '${archivePathUri}'.")
}
new Path(archivePathUri)
}
[3/4] flink git commit: [FLINK-7643] [core] Rework FileSystem loading
to use factories
Posted by se...@apache.org.
http://git-wip-us.apache.org/repos/asf/flink/blob/536675b0/flink-runtime/src/main/java/org/apache/flink/runtime/fs/maprfs/MapRFileSystem.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/fs/maprfs/MapRFileSystem.java b/flink-runtime/src/main/java/org/apache/flink/runtime/fs/maprfs/MapRFileSystem.java
index 275e492..9e12f96 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/fs/maprfs/MapRFileSystem.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/fs/maprfs/MapRFileSystem.java
@@ -37,16 +37,20 @@ import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
import java.net.URI;
import java.util.ArrayList;
import java.util.List;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
/**
* Concrete implementation of the {@link FileSystem} base class for the MapR
* file system. The class contains MapR specific code to initialize the
* connection to the file system. Apart from that, we code mainly reuses the
* existing HDFS wrapper code.
*/
+@SuppressWarnings("unused") // is only instantiated via reflection
public final class MapRFileSystem extends FileSystem {
/**
@@ -77,21 +81,12 @@ public final class MapRFileSystem extends FileSystem {
*/
private static final String MAPR_CLUSTER_CONF_FILE = "/conf/mapr-clusters.conf";
- /**
- * A Hadoop configuration object used during the file system initialization.
- */
- private final org.apache.hadoop.conf.Configuration conf = new org.apache.hadoop.conf.Configuration();
-
- /**
- * The MapR class containing the implementation of the Hadoop HDFS
- * interface.
- */
- private final Class<? extends org.apache.hadoop.fs.FileSystem> fsClass;
+ // ------------------------------------------------------------------------
/**
* The MapR implementation of the Hadoop HDFS interface.
*/
- private org.apache.hadoop.fs.FileSystem fs;
+ private final org.apache.hadoop.fs.FileSystem fs;
/**
* Creates a new MapRFileSystem object to access the MapR file system.
@@ -99,59 +94,36 @@ public final class MapRFileSystem extends FileSystem {
* @throws IOException
* throw if the required MapR classes cannot be found
*/
- @SuppressWarnings("unchecked")
- public MapRFileSystem() throws IOException {
+ public MapRFileSystem(URI fsURI) throws IOException {
+ checkNotNull(fsURI, "fsURI");
- if (LOG.isDebugEnabled()) {
- LOG.debug(String.format(
- "Trying to load class %s to access the MapR file system",
- MAPR_FS_IMPL_CLASS));
- }
+ LOG.debug("Trying to load class {} to access the MapR file system", MAPR_FS_IMPL_CLASS);
+ final Class<? extends org.apache.hadoop.fs.FileSystem> fsClass;
try {
- this.fsClass = (Class<? extends org.apache.hadoop.fs.FileSystem>) Class
- .forName(MAPR_FS_IMPL_CLASS);
- } catch (Exception e) {
+ fsClass = Class.forName(MAPR_FS_IMPL_CLASS).asSubclass(org.apache.hadoop.fs.FileSystem.class);
+ }
+ catch (Exception e) {
throw new IOException(
- String.format(
- "Cannot find class %s, probably the runtime was not compiled against the MapR Hadoop libraries",
+ String.format("Cannot load MapR File System class '%s'. " +
+ "Please check that the MapR Hadoop libraries are in the classpath.",
MAPR_FS_IMPL_CLASS), e);
}
- }
-
- @Override
- public Path getWorkingDirectory() {
-
- return new Path(this.fs.getWorkingDirectory().toUri());
- }
-
- public Path getHomeDirectory() {
- return new Path(this.fs.getHomeDirectory().toUri());
- }
-
- @Override
- public URI getUri() {
- return this.fs.getUri();
- }
+ LOG.info("Initializing MapR file system for URI {}", fsURI);
- @Override
- public void initialize(final URI path) throws IOException {
+ final org.apache.hadoop.conf.Configuration conf = new org.apache.hadoop.conf.Configuration();
+ final org.apache.hadoop.fs.FileSystem fs;
- if (LOG.isInfoEnabled()) {
- LOG.info(String.format("Initializing MapR file system for path %s",
- path.toString()));
- }
-
- final String authority = path.getAuthority();
+ final String authority = fsURI.getAuthority();
if (authority == null || authority.isEmpty()) {
- // Use the default constructor to instantiate MapR file system
- // object
+ // Use the default constructor to instantiate MapR file system object
try {
- this.fs = this.fsClass.newInstance();
- } catch (Exception e) {
+ fs = fsClass.newInstance();
+ }
+ catch (Exception e) {
throw new IOException(e);
}
} else {
@@ -161,110 +133,47 @@ public final class MapRFileSystem extends FileSystem {
final String[] cldbLocations = getCLDBLocations(authority);
// Find the appropriate constructor
- final Constructor<? extends org.apache.hadoop.fs.FileSystem> constructor;
try {
- constructor = this.fsClass.getConstructor(String.class,
- String[].class);
- } catch (NoSuchMethodException e) {
- throw new IOException(e);
- }
+ final Constructor<? extends org.apache.hadoop.fs.FileSystem> constructor =
+ fsClass.getConstructor(String.class, String[].class);
- // Instantiate the file system object
- try {
- this.fs = constructor.newInstance(authority, cldbLocations);
- } catch (Exception e) {
+ fs = constructor.newInstance(authority, cldbLocations);
+ }
+ catch (InvocationTargetException e) {
+ if (e.getTargetException() instanceof IOException) {
+ throw (IOException) e.getTargetException();
+ } else {
+ throw new IOException(e.getTargetException());
+ }
+ }
+ catch (Exception e) {
throw new IOException(e);
}
}
- this.fs.initialize(path, this.conf);
- }
-
- /**
- * Retrieves the CLDB locations for the given MapR cluster name.
- *
- * @param authority
- * the name of the MapR cluster
- * @return a list of CLDB locations
- * @throws IOException
- * thrown if the CLDB locations for the given MapR cluster name
- * cannot be determined
- */
- private static String[] getCLDBLocations(final String authority)
- throws IOException {
+ // now initialize the Hadoop File System object
+ fs.initialize(fsURI, conf);
- // Determine the MapR home
- String maprHome = System.getenv(MAPR_HOME_ENV);
- if (maprHome == null) {
- maprHome = DEFAULT_MAPR_HOME;
- }
-
- final File maprClusterConf = new File(maprHome, MAPR_CLUSTER_CONF_FILE);
-
- if (LOG.isDebugEnabled()) {
- LOG.debug(String.format(
- "Trying to retrieve MapR cluster configuration from %s",
- maprClusterConf));
- }
-
- // Read the cluster configuration file, format is specified at
- // http://doc.mapr.com/display/MapR/mapr-clusters.conf
- BufferedReader br = null;
- try {
- br = new BufferedReader(new FileReader(maprClusterConf));
-
- String line;
- while ((line = br.readLine()) != null) {
-
- // Normalize the string
- line = line.trim();
- line = line.replace('\t', ' ');
-
- final String[] fields = line.split(" ");
- if (fields == null) {
- continue;
- }
-
- if (fields.length < 1) {
- continue;
- }
-
- final String clusterName = fields[0];
-
- if (!clusterName.equals(authority)) {
- continue;
- }
-
- final List<String> cldbLocations = new ArrayList<String>();
-
- for (int i = 1; i < fields.length; ++i) {
-
- // Make sure this is not a key-value pair MapR recently
- // introduced in the file format along with their security
- // features.
- if (!fields[i].isEmpty() && !fields[i].contains("=")) {
- cldbLocations.add(fields[i]);
- }
- }
+ // all good as it seems
+ this.fs = fs;
+ }
- if (cldbLocations.isEmpty()) {
- throw new IOException(
- String.format(
- "%s contains entry for cluster %s but no CLDB locations.",
- maprClusterConf, authority));
- }
+ // ------------------------------------------------------------------------
+ // file system methods
+ // ------------------------------------------------------------------------
- return cldbLocations.toArray(new String[0]);
- }
+ @Override
+ public Path getWorkingDirectory() {
+ return new Path(this.fs.getWorkingDirectory().toUri());
+ }
- } finally {
- if (br != null) {
- br.close();
- }
- }
+ public Path getHomeDirectory() {
+ return new Path(this.fs.getHomeDirectory().toUri());
+ }
- throw new IOException(String.format(
- "Unable to find CLDB locations for cluster %s", authority));
+ @Override
+ public URI getUri() {
+ return this.fs.getUri();
}
@Override
@@ -315,6 +224,7 @@ public final class MapRFileSystem extends FileSystem {
return new HadoopDataInputStream(fdis);
}
+ @SuppressWarnings("deprecation")
@Override
public FSDataOutputStream create(final Path f, final boolean overwrite,
final int bufferSize, final short replication, final long blockSize)
@@ -376,13 +286,92 @@ public final class MapRFileSystem extends FileSystem {
@SuppressWarnings("deprecation")
@Override
public long getDefaultBlockSize() {
-
return this.fs.getDefaultBlockSize();
}
@Override
public boolean isDistributedFS() {
-
return true;
}
+
+ // ------------------------------------------------------------------------
+ // Utilities
+ // ------------------------------------------------------------------------
+
+ /**
+ * Retrieves the CLDB locations for the given MapR cluster name.
+ *
+ * @param authority
+ * the name of the MapR cluster
+ * @return a list of CLDB locations
+ * @throws IOException
+ * thrown if the CLDB locations for the given MapR cluster name
+ * cannot be determined
+ */
+ private static String[] getCLDBLocations(final String authority) throws IOException {
+
+ // Determine the MapR home
+ String maprHome = System.getenv(MAPR_HOME_ENV);
+ if (maprHome == null) {
+ maprHome = DEFAULT_MAPR_HOME;
+ }
+
+ final File maprClusterConf = new File(maprHome, MAPR_CLUSTER_CONF_FILE);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(String.format(
+ "Trying to retrieve MapR cluster configuration from %s",
+ maprClusterConf));
+ }
+
+ // Read the cluster configuration file, format is specified at
+ // http://doc.mapr.com/display/MapR/mapr-clusters.conf
+
+ try (BufferedReader br = new BufferedReader(new FileReader(maprClusterConf))) {
+
+ String line;
+ while ((line = br.readLine()) != null) {
+
+ // Normalize the string
+ line = line.trim();
+ line = line.replace('\t', ' ');
+
+ final String[] fields = line.split(" ");
+ if (fields.length < 1) {
+ continue;
+ }
+
+ final String clusterName = fields[0];
+
+ if (!clusterName.equals(authority)) {
+ continue;
+ }
+
+ final List<String> cldbLocations = new ArrayList<>();
+
+ for (int i = 1; i < fields.length; ++i) {
+
+ // Make sure this is not a key-value pair MapR recently
+ // introduced in the file format along with their security
+ // features.
+ if (!fields[i].isEmpty() && !fields[i].contains("=")) {
+ cldbLocations.add(fields[i]);
+ }
+ }
+
+ if (cldbLocations.isEmpty()) {
+ throw new IOException(
+ String.format(
+ "%s contains entry for cluster %s but no CLDB locations.",
+ maprClusterConf, authority));
+ }
+
+ return cldbLocations.toArray(new String[cldbLocations.size()]);
+ }
+
+ }
+
+ throw new IOException(String.format(
+ "Unable to find CLDB locations for cluster %s", authority));
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/536675b0/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
index 56c79bb..85a6aed 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
@@ -2386,7 +2386,7 @@ object JobManager {
val configuration = GlobalConfiguration.loadConfiguration(configDir)
try {
- FileSystem.setDefaultScheme(configuration)
+ FileSystem.initialize(configuration)
}
catch {
case e: IOException => {
http://git-wip-us.apache.org/repos/asf/flink/blob/536675b0/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
index 8d1f6f7..558388c 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
@@ -1617,7 +1617,7 @@ object TaskManager {
}
try {
- FileSystem.setDefaultScheme(conf)
+ FileSystem.initialize(conf)
}
catch {
case e: IOException => {
http://git-wip-us.apache.org/repos/asf/flink/blob/536675b0/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerConfigurationTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerConfigurationTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerConfigurationTest.java
index e73c684..5f2de33 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerConfigurationTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerConfigurationTest.java
@@ -29,6 +29,7 @@ import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.runtime.concurrent.Executors;
import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
import org.apache.flink.runtime.highavailability.HighAvailabilityServicesUtils;
+import org.apache.flink.util.IOUtils;
import org.junit.Rule;
import org.junit.Test;
@@ -38,8 +39,9 @@ import scala.Tuple2;
import java.io.File;
import java.io.IOException;
import java.io.PrintWriter;
-import java.lang.reflect.Field;
-import java.net.*;
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import java.net.URI;
import java.util.Iterator;
import static org.junit.Assert.*;
@@ -78,10 +80,6 @@ public class TaskManagerConfigurationTest {
// validate the configured test host name
assertEquals(TEST_HOST_NAME, address._1());
- }
- catch (Exception e) {
- e.printStackTrace();
- fail(e.getMessage());
} finally {
highAvailabilityServices.closeAndCleanupAllData();
}
@@ -164,16 +162,11 @@ public class TaskManagerConfigurationTest {
String[] args = new String[] {"--configDir:" + tmpDir};
TaskManager.parseArgsAndLoadConfig(args);
- Field f = FileSystem.class.getDeclaredField("defaultScheme");
- f.setAccessible(true);
- URI scheme = (URI) f.get(null);
-
- assertEquals("Default Filesystem Scheme not configured.", scheme, defaultFS);
- } finally {
- // reset default FS scheme:
- Field f = FileSystem.class.getDeclaredField("defaultScheme");
- f.setAccessible(true);
- f.set(null, null);
+ assertEquals(defaultFS, FileSystem.getDefaultFsUri());
+ }
+ finally {
+ // reset FS settings
+ FileSystem.initialize(new Configuration());
}
}
@@ -205,18 +198,9 @@ public class TaskManagerConfigurationTest {
try {
assertNotNull(TaskManager.selectNetworkInterfaceAndPortRange(config, highAvailabilityServices)._1());
}
- catch (Exception e) {
- e.printStackTrace();
- fail(e.getMessage());
- }
finally {
highAvailabilityServices.closeAndCleanupAllData();
-
- try {
- server.close();
- } catch (IOException e) {
- // ignore shutdown errors
- }
+ IOUtils.closeQuietly(server);
}
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/536675b0/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java b/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java
index de78393..8ecc371 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java
@@ -611,10 +611,10 @@ public abstract class AbstractYarnClusterDescriptor implements ClusterDescriptor
YarnClientApplication yarnApplication,
ClusterSpecification clusterSpecification) throws Exception {
- // ------------------ Set default file system scheme -------------------------
+ // ------------------ Initialize the file systems -------------------------
try {
- org.apache.flink.core.fs.FileSystem.setDefaultScheme(flinkConfiguration);
+ org.apache.flink.core.fs.FileSystem.initialize(flinkConfiguration);
} catch (IOException e) {
throw new IOException("Error while setting the default " +
"filesystem scheme from configuration.", e);
http://git-wip-us.apache.org/repos/asf/flink/blob/536675b0/flink-yarn/src/main/java/org/apache/flink/yarn/YarnTaskExecutorRunner.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnTaskExecutorRunner.java b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnTaskExecutorRunner.java
index 25d46fb..cd0053b 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnTaskExecutorRunner.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnTaskExecutorRunner.java
@@ -100,7 +100,7 @@ public class YarnTaskExecutorRunner {
LOG.info("TM: remote keytab principal obtained {}", remoteKeytabPrincipal);
final Configuration configuration = GlobalConfiguration.loadConfiguration(currDir);
- FileSystem.setDefaultScheme(configuration);
+ FileSystem.initialize(configuration);
// configure local directory
String flinkTempDirs = configuration.getString(ConfigConstants.TASK_MANAGER_TMP_DIR_KEY, null);
http://git-wip-us.apache.org/repos/asf/flink/blob/536675b0/flink-yarn/src/main/java/org/apache/flink/yarn/highavailability/YarnHighAvailabilityServices.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/highavailability/YarnHighAvailabilityServices.java b/flink-yarn/src/main/java/org/apache/flink/yarn/highavailability/YarnHighAvailabilityServices.java
index d94921e..0cf9dc4 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/highavailability/YarnHighAvailabilityServices.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/highavailability/YarnHighAvailabilityServices.java
@@ -142,7 +142,7 @@ public abstract class YarnHighAvailabilityServices implements HighAvailabilitySe
throw new IOException("Cannot instantiate YARN's Hadoop file system for " + fsUri, e);
}
- this.flinkFileSystem = new HadoopFileSystem(hadoopConf, hadoopFileSystem);
+ this.flinkFileSystem = new HadoopFileSystem(hadoopFileSystem);
this.workingDirectory = new Path(hadoopFileSystem.getWorkingDirectory().toUri());
this.haDataDirectory = new Path(workingDirectory, FLINK_RECOVERY_DATA_DIR);
[4/4] flink git commit: [FLINK-7643] [core] Rework FileSystem loading
to use factories
Posted by se...@apache.org.
[FLINK-7643] [core] Rework FileSystem loading to use factories
This makes sure that configurations are loaded once and file system instances are
properly reused by scheme and authority.
This also factors out a lot of the special treatment of Hadoop file systems and simply
makes the Hadoop File System factory the default fallback factory.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/536675b0
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/536675b0
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/536675b0
Branch: refs/heads/master
Commit: 536675b03a5050fda9c3e1fd403818cb50dcc6ff
Parents: a5ef09b
Author: Stephan Ewen <se...@apache.org>
Authored: Mon Oct 2 16:25:18 2017 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Fri Oct 6 11:16:10 2017 +0200
----------------------------------------------------------------------
.../org/apache/flink/client/CliFrontend.java | 2 +-
flink-core/pom.xml | 3 +
.../org/apache/flink/core/fs/FileSystem.java | 304 +++++++---------
.../apache/flink/core/fs/FileSystemFactory.java | 56 +++
.../flink/core/fs/HadoopFileSystemWrapper.java | 32 --
.../core/fs/SafetyNetWrapperFileSystem.java | 5 -
.../UnsupportedFileSystemSchemeException.java | 48 +++
.../HadoopFileSystemFactoryLoader.java | 81 +++++
.../fs/factories/LocalFileSystemFactory.java | 44 +++
.../flink/core/fs/factories/MapRFsFactory.java | 75 ++++
.../fs/factories/UnsupportedSchemeFactory.java | 65 ++++
.../flink/core/fs/local/LocalFileSystem.java | 21 +-
.../FilesystemSchemeConfigTest.java | 142 +++-----
.../apache/flink/testutils/TestFileSystem.java | 33 +-
.../entrypoint/MesosTaskExecutorRunner.java | 7 +-
.../MesosApplicationMasterRunner.java | 7 +-
.../MesosTaskManagerRunner.java | 7 +-
.../runtime/entrypoint/ClusterEntrypoint.java | 6 +-
.../flink/runtime/fs/hdfs/HadoopFileSystem.java | 347 ++-----------------
.../flink/runtime/fs/hdfs/HadoopFsFactory.java | 167 +++++++++
.../flink/runtime/fs/maprfs/MapRFileSystem.java | 281 ++++++++-------
.../flink/runtime/jobmanager/JobManager.scala | 2 +-
.../flink/runtime/taskmanager/TaskManager.scala | 2 +-
.../TaskManagerConfigurationTest.java | 36 +-
.../yarn/AbstractYarnClusterDescriptor.java | 4 +-
.../flink/yarn/YarnTaskExecutorRunner.java | 2 +-
.../YarnHighAvailabilityServices.java | 2 +-
27 files changed, 959 insertions(+), 822 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/536675b0/flink-clients/src/main/java/org/apache/flink/client/CliFrontend.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/CliFrontend.java b/flink-clients/src/main/java/org/apache/flink/client/CliFrontend.java
index 52c655e..9d1f52e 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/CliFrontend.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/CliFrontend.java
@@ -179,7 +179,7 @@ public class CliFrontend {
this.config = GlobalConfiguration.loadConfiguration(configDirectory.getAbsolutePath());
try {
- FileSystem.setDefaultScheme(config);
+ FileSystem.initialize(config);
} catch (IOException e) {
throw new Exception("Error while setting the default " +
"filesystem scheme from configuration.", e);
http://git-wip-us.apache.org/repos/asf/flink/blob/536675b0/flink-core/pom.xml
----------------------------------------------------------------------
diff --git a/flink-core/pom.xml b/flink-core/pom.xml
index 7039e48..5b97273 100644
--- a/flink-core/pom.xml
+++ b/flink-core/pom.xml
@@ -195,6 +195,9 @@ under the License.
<excludes combine.children="append">
<exclude>org.apache.flink.api.common.ExecutionConfig#CONFIG_KEY</exclude>
<exclude>org.apache.flink.core.fs.FileSystem\$FSKey</exclude>
+ <exclude>org.apache.flink.core.fs.FileSystem#initialize(java.net.URI)</exclude>
+ <exclude>org.apache.flink.core.fs.FileSystem#isFlinkSupportedScheme(java.lang.String)</exclude>
+ <exclude>org.apache.flink.core.fs.FileSystem#setDefaultScheme(org.apache.flink.configuration.Configuration)</exclude>
<exclude>org.apache.flink.api.java.typeutils.WritableTypeInfo</exclude>
<!-- Breaking changes between 1.1 and 1.2.
We ignore these changes because these are low-level, internal runtime configuration parameters -->
http://git-wip-us.apache.org/repos/asf/flink/blob/536675b0/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 672ebbb..6c18735 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
@@ -29,13 +29,16 @@ import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.Public;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.IllegalConfigurationException;
+import org.apache.flink.core.fs.factories.HadoopFileSystemFactoryLoader;
+import org.apache.flink.core.fs.factories.MapRFsFactory;
import org.apache.flink.core.fs.local.LocalFileSystem;
+import org.apache.flink.core.fs.factories.LocalFileSystemFactory;
import javax.annotation.Nullable;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
-import java.lang.reflect.Constructor;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.HashMap;
@@ -194,19 +197,9 @@ public abstract class FileSystem {
}
// ------------------------------------------------------------------------
- // File System Implementation Classes
- // ------------------------------------------------------------------------
-
- private static final String HADOOP_WRAPPER_FILESYSTEM_CLASS = "org.apache.flink.runtime.fs.hdfs.HadoopFileSystem";
-
- private static final String MAPR_FILESYSTEM_CLASS = "org.apache.flink.runtime.fs.maprfs.MapRFileSystem";
-
- private static final String HADOOP_WRAPPER_SCHEME = "hdwrapper";
-
- // ------------------------------------------------------------------------
/** This lock guards the methods {@link #initOutPathLocalFS(Path, WriteMode, boolean)} and
- * {@link #initOutPathDistFS(Path, WriteMode, boolean)} which are otherwise susceptible to races */
+ * {@link #initOutPathDistFS(Path, WriteMode, boolean)} which are otherwise susceptible to races. */
private static final ReentrantLock OUTPUT_DIRECTORY_INIT_LOCK = new ReentrantLock(true);
/** Object used to protect calls to specific methods.*/
@@ -215,58 +208,67 @@ public abstract class FileSystem {
/** Cache for file systems, by scheme + authority. */
private static final Map<FSKey, FileSystem> CACHE = new HashMap<>();
- /** Mapping of file system schemes to the corresponding implementations */
- private static final Map<String, String> FSDIRECTORY = new HashMap<>();
+ /** Mapping of file system schemes to the corresponding implementation factories. */
+ private static final Map<String, FileSystemFactory> FS_FACTORIES = new HashMap<>();
- /** The local file system. Needs to be lazily initialized to avoid that some JVMs deadlock
- * on static subclass initialization. */
- private static LocalFileSystem LOCAL_FS;
+ /** The default factory that is used when no scheme matches. */
+ private static final FileSystemFactory FALLBACK_FACTORY = HadoopFileSystemFactoryLoader.loadFactory();
/** The default filesystem scheme to be used, configured during process-wide initialization.
- * This value defaults to the local file systems scheme {@code 'file:///'} or
- * {@code 'file:/'}. */
- private static URI defaultScheme;
+ * This value defaults to the local file systems scheme {@code 'file:///'} or {@code 'file:/'}. */
+ private static URI DEFAULT_SCHEME;
// ------------------------------------------------------------------------
// Initialization
// ------------------------------------------------------------------------
static {
- FSDIRECTORY.put("hdfs", HADOOP_WRAPPER_FILESYSTEM_CLASS);
- FSDIRECTORY.put("maprfs", MAPR_FILESYSTEM_CLASS);
- FSDIRECTORY.put("file", LocalFileSystem.class.getName());
+ FS_FACTORIES.put("file", new LocalFileSystemFactory());
+ FS_FACTORIES.put("maprfs", new MapRFsFactory());
}
/**
- * <p>
- * Sets the default filesystem scheme based on the user-specified configuration parameter
- * <code>fs.default-scheme</code>. By default this is set to <code>file:///</code>
- * (see {@link ConfigConstants#FILESYSTEM_SCHEME} and
- * {@link ConfigConstants#DEFAULT_FILESYSTEM_SCHEME}),
- * and the local filesystem is used.
- * <p>
- * As an example, if set to <code>hdfs://localhost:9000/</code>, then an HDFS deployment
- * with the namenode being on the local node and listening to port 9000 is going to be used.
- * In this case, a file path specified as <code>/user/USERNAME/in.txt</code>
- * is going to be transformed into <code>hdfs://localhost:9000/user/USERNAME/in.txt</code>. By
- * default this is set to <code>file:///</code> which points to the local filesystem.
+ * Initializes the shared file system settings.
+ *
+ * <p>The given configuration is passed to each file system factory to initialize the respective
+ * file systems. Because the configuration of file systems may be different subsequent to the call
+ * of this method, this method clears the file system instance cache.
+ *
+ * <p>This method also reads the default file system URI from the configuration key
+ * {@link ConfigConstants#FILESYSTEM_SCHEME}. All calls to {@link FileSystem#get(URI)} where
+ * the URI has no scheme will be interpreted as relative to that URI.
+ * As an example, assume the default file system URI is set to {@code 'hdfs://localhost:9000/'}.
+ * A file path of {@code '/user/USERNAME/in.txt'} is interpreted as
+ * {@code 'hdfs://localhost:9000/user/USERNAME/in.txt'}.
+ *
* @param config the configuration from where to fetch the parameter.
*/
- public static void setDefaultScheme(Configuration config) throws IOException {
+ public static void initialize(Configuration config) throws IOException, IllegalConfigurationException {
LOCK.lock();
try {
- if (defaultScheme == null) {
- final String stringifiedUri = config.getString(ConfigConstants.FILESYSTEM_SCHEME, null);
- if (stringifiedUri == null) {
- defaultScheme = LocalFileSystem.getLocalFsURI();
+ // make sure file systems are re-instantiated after re-configuration
+ CACHE.clear();
+
+ // configure all file system factories
+ for (FileSystemFactory factory : FS_FACTORIES.values()) {
+ factory.configure(config);
+ }
+
+ // configure the default (fallback) factory
+ FALLBACK_FACTORY.configure(config);
+
+ // also read the default file system scheme
+ final String stringifiedUri = config.getString(ConfigConstants.FILESYSTEM_SCHEME, null);
+ if (stringifiedUri == null) {
+ DEFAULT_SCHEME = null;
+ }
+ else {
+ try {
+ DEFAULT_SCHEME = new URI(stringifiedUri);
}
- else {
- try {
- defaultScheme = new URI(stringifiedUri);
- } catch (URISyntaxException e) {
- throw new IOException("The URI used to set the default filesystem " +
- "scheme ('" + stringifiedUri + "') is not valid.");
- }
+ catch (URISyntaxException e) {
+ throw new IllegalConfigurationException("The default file system scheme ('" +
+ ConfigConstants.FILESYSTEM_SCHEME + "') is invalid: " + stringifiedUri, e);
}
}
}
@@ -285,52 +287,69 @@ public abstract class FileSystem {
* @return a reference to the {@link FileSystem} instance for accessing the local file system.
*/
public static FileSystem getLocalFileSystem() {
- LOCK.lock();
- try {
- if (LOCAL_FS == null) {
- LOCAL_FS = new LocalFileSystem();
- }
- return FileSystemSafetyNet.wrapWithSafetyNetWhenActivated(LOCAL_FS);
- } finally {
- LOCK.unlock();
- }
+ return FileSystemSafetyNet.wrapWithSafetyNetWhenActivated(LocalFileSystem.getSharedInstance());
+ }
+
+ /**
+ * Returns a reference to the {@link FileSystem} instance for accessing the
+ * file system identified by the given {@link URI}.
+ *
+ * @param uri
+ * the {@link URI} identifying the file system
+ * @return a reference to the {@link FileSystem} instance for accessing the file system identified by the given
+ * {@link URI}.
+ * @throws IOException
+ * thrown if a reference to the file system instance could not be obtained
+ */
+ public static FileSystem get(URI uri) throws IOException {
+ return FileSystemSafetyNet.wrapWithSafetyNetWhenActivated(getUnguardedFileSystem(uri));
}
@Internal
- public static FileSystem getUnguardedFileSystem(URI uri) throws IOException {
- final URI asked = uri;
+ public static FileSystem getUnguardedFileSystem(final URI fsUri) throws IOException {
+ checkNotNull(fsUri, "file system URI");
LOCK.lock();
try {
+ final URI uri;
- 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);
+ if (fsUri.getScheme() != null) {
+ uri = fsUri;
+ }
+ else {
+ // Apply the default fs scheme
+ final URI defaultUri = getDefaultFsUri();
+ URI rewrittenUri = 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);
+ try {
+ rewrittenUri = new URI(defaultUri.getScheme(), null, defaultUri.getHost(),
+ defaultUri.getPort(), fsUri.getPath(), null, null);
+ }
+ catch (URISyntaxException e) {
+ // for local URIs, we make one more try to repair the path by making it absolute
+ if (defaultUri.getScheme().equals("file")) {
+ try {
+ rewrittenUri = new URI(
+ "file", null,
+ new Path(new File(fsUri.getPath()).getAbsolutePath()).toUri().getPath(),
+ null);
+ } catch (URISyntaxException ignored) {
+ // could not help it...
}
- } 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 (rewrittenUri != null) {
+ uri = rewrittenUri;
+ }
+ else {
+ throw new IOException("The file system URI '" + fsUri +
+ "' declares no scheme and cannot be interpreted relative to the default file system URI ("
+ + defaultUri + ").");
+ }
}
+ // print a helpful pointer for malformed local URIs (happens a lot to new users)
if (uri.getScheme().equals("file") && uri.getAuthority() != null && !uri.getAuthority().isEmpty()) {
String supposedUri = "file:///" + uri.getAuthority() + uri.getPath();
@@ -341,49 +360,33 @@ public abstract class FileSystem {
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);
+ {
+ FileSystem cached = CACHE.get(key);
+ if (cached != null) {
+ return cached;
+ }
}
// Try to create a new file system
final FileSystem fs;
+ final FileSystemFactory factory = FS_FACTORIES.get(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) {
- // 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() + "'.");
+ if (factory != null) {
+ fs = factory.create(uri);
+ }
+ else {
+ try {
+ fs = FALLBACK_FACTORY.create(uri);
}
- } 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);
+ catch (UnsupportedFileSystemSchemeException e) {
+ throw new UnsupportedFileSystemSchemeException(
+ "Could not find a file system implementation for scheme '" + uri.getScheme() +
+ "'. The scheme is not directly supported by Flink and no Hadoop file " +
+ "system to support this scheme could be loaded.", e);
}
- // Initialize new file system object
- fs.initialize(uri);
-
- // Add new file system object to cache
- CACHE.put(key, fs);
}
+ CACHE.put(key, fs);
return fs;
}
finally {
@@ -392,58 +395,19 @@ public abstract class FileSystem {
}
/**
- * Returns a reference to the {@link FileSystem} instance for accessing the
- * file system identified by the given {@link URI}.
+ * Gets the default file system URI that is used for paths and file systems
+ * that do not specify and explicit scheme.
*
- * @param uri
- * the {@link URI} identifying the file system
- * @return a reference to the {@link FileSystem} instance for accessing the file system identified by the given
- * {@link URI}.
- * @throws IOException
- * thrown if a reference to the file system instance could not be obtained
+ * <p>As an example, assume the default file system URI is set to {@code 'hdfs://someserver:9000/'}.
+ * A file path of {@code '/user/USERNAME/in.txt'} is interpreted as
+ * {@code 'hdfs://someserver:9000/user/USERNAME/in.txt'}.
+ *
+ * @return The default file system URI
*/
- public static FileSystem get(URI uri) throws IOException {
- return FileSystemSafetyNet.wrapWithSafetyNetWhenActivated(getUnguardedFileSystem(uri));
- }
-
- //Class must implement Hadoop FileSystem interface. The class is not avaiable in 'flink-core'.
- private static FileSystem instantiateHadoopFileSystemWrapper(Class<?> wrappedFileSystem) throws IOException {
- try {
- Class<? extends FileSystem> fsClass = getFileSystemByName(HADOOP_WRAPPER_FILESYSTEM_CLASS);
- Constructor<? extends FileSystem> fsClassCtor = fsClass.getConstructor(Class.class);
- return fsClassCtor.newInstance(wrappedFileSystem);
- } catch (Throwable e) {
- throw new IOException("Error loading Hadoop FS wrapper", e);
- }
- }
-
- private static FileSystem instantiateFileSystem(String className) throws IOException {
- try {
- Class<? extends FileSystem> fsClass = getFileSystemByName(className);
- return fsClass.newInstance();
- }
- catch (ClassNotFoundException e) {
- throw new IOException("Could not load file system class '" + className + '\'', e);
- }
- catch (InstantiationException | IllegalAccessException e) {
- throw new IOException("Could not instantiate file system class: " + e.getMessage(), e);
- }
- }
-
- private static HadoopFileSystemWrapper hadoopWrapper;
-
- private static Class<?> getHadoopWrapperClassNameForFileSystem(String scheme) {
- if (hadoopWrapper == null) {
- try {
- hadoopWrapper = (HadoopFileSystemWrapper) instantiateHadoopFileSystemWrapper(null);
- } catch (IOException e) {
- throw new RuntimeException("Error creating new Hadoop wrapper", e);
- }
- }
- return hadoopWrapper.getHadoopWrapperClassNameForFileSystem(scheme);
+ public static URI getDefaultFsUri() {
+ return DEFAULT_SCHEME != null ? DEFAULT_SCHEME : LocalFileSystem.getLocalFsURI();
}
-
// ------------------------------------------------------------------------
// File System Methods
// ------------------------------------------------------------------------
@@ -470,14 +434,6 @@ public abstract class FileSystem {
public abstract URI getUri();
/**
- * Called after a new FileSystem instance is constructed.
- *
- * @param name
- * a {@link URI} whose authority section names the host, port, etc. for this file system
- */
- public abstract void initialize(URI name) throws IOException;
-
- /**
* Return a file status object that represents the path.
*
* @param f
@@ -936,14 +892,6 @@ public abstract class FileSystem {
}
}
- // ------------------------------------------------------------------------
- // utilities
- // ------------------------------------------------------------------------
-
- private static Class<? extends FileSystem> getFileSystemByName(String className) throws ClassNotFoundException {
- return Class.forName(className, true, FileSystem.class.getClassLoader()).asSubclass(FileSystem.class);
- }
-
/**
* An identifier of a file system, via its scheme and its authority.
*/
@@ -991,7 +939,7 @@ public abstract class FileSystem {
@Override
public String toString() {
- return scheme + "://" + authority;
+ return scheme + "://" + (authority != null ? authority : "");
}
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/536675b0/flink-core/src/main/java/org/apache/flink/core/fs/FileSystemFactory.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/core/fs/FileSystemFactory.java b/flink-core/src/main/java/org/apache/flink/core/fs/FileSystemFactory.java
new file mode 100644
index 0000000..503f21f
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/core/fs/FileSystemFactory.java
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.core.fs;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.Configuration;
+
+import java.io.IOException;
+import java.net.URI;
+
+/**
+ * A factory to create file systems.
+ *
+ * <p>The factory is typically configured via {@link #configure(Configuration)} before
+ * creating file systems via {@link #create(URI)}.
+ */
+@PublicEvolving
+public interface FileSystemFactory {
+
+ /**
+ * Applies the given configuration to this factory. All future file system
+ * instantiations via {@link #create(URI)} should take the configuration into
+ * account.
+ *
+ * @param config The configuration to apply.
+ */
+ void configure(Configuration config);
+
+ /**
+ * Creates a new file system for the given file system URI.
+ * The URI describes the type of file system (via its scheme) and optionally the
+ * authority (for example the host) of the file system.
+ *
+ * @param fsUri The URI that describes the file system.
+ * @return A new instance of the specified file system.
+ *
+ * @throws IOException Thrown if the file system could not be instantiated.
+ */
+ FileSystem create(URI fsUri) throws IOException;
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/536675b0/flink-core/src/main/java/org/apache/flink/core/fs/HadoopFileSystemWrapper.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/core/fs/HadoopFileSystemWrapper.java b/flink-core/src/main/java/org/apache/flink/core/fs/HadoopFileSystemWrapper.java
deleted file mode 100644
index 5a48e68..0000000
--- a/flink-core/src/main/java/org/apache/flink/core/fs/HadoopFileSystemWrapper.java
+++ /dev/null
@@ -1,32 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.core.fs;
-
-import org.apache.flink.annotation.PublicEvolving;
-
-@PublicEvolving
-public interface HadoopFileSystemWrapper {
-
- /**
- * Test whether the HadoopWrapper can wrap the given file system scheme.
- *
- * @param scheme The scheme of the file system.
- * @return The class implementing the file system.
- */
- public Class<?> getHadoopWrapperClassNameForFileSystem(String scheme);
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/536675b0/flink-core/src/main/java/org/apache/flink/core/fs/SafetyNetWrapperFileSystem.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/core/fs/SafetyNetWrapperFileSystem.java b/flink-core/src/main/java/org/apache/flink/core/fs/SafetyNetWrapperFileSystem.java
index 1dacafd..a1167dd 100644
--- a/flink-core/src/main/java/org/apache/flink/core/fs/SafetyNetWrapperFileSystem.java
+++ b/flink-core/src/main/java/org/apache/flink/core/fs/SafetyNetWrapperFileSystem.java
@@ -60,11 +60,6 @@ public class SafetyNetWrapperFileSystem extends FileSystem implements WrappingPr
}
@Override
- public void initialize(URI name) throws IOException {
- unsafeFileSystem.initialize(name);
- }
-
- @Override
public FileStatus getFileStatus(Path f) throws IOException {
return unsafeFileSystem.getFileStatus(f);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/536675b0/flink-core/src/main/java/org/apache/flink/core/fs/UnsupportedFileSystemSchemeException.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/core/fs/UnsupportedFileSystemSchemeException.java b/flink-core/src/main/java/org/apache/flink/core/fs/UnsupportedFileSystemSchemeException.java
new file mode 100644
index 0000000..1dbfcaf
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/core/fs/UnsupportedFileSystemSchemeException.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.core.fs;
+
+import java.io.IOException;
+
+/**
+ * An exception to indicate that a specific file system scheme is not supported.
+ */
+public class UnsupportedFileSystemSchemeException extends IOException {
+
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * Creates a new exception with the given message.
+ *
+ * @param message The exception message
+ */
+ public UnsupportedFileSystemSchemeException(String message) {
+ super(message);
+ }
+
+ /**
+ * Creates a new exception with the given message and cause.
+ *
+ * @param message The exception message
+ * @param cause The exception cause
+ */
+ public UnsupportedFileSystemSchemeException(String message, Throwable cause) {
+ super(message, cause);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/536675b0/flink-core/src/main/java/org/apache/flink/core/fs/factories/HadoopFileSystemFactoryLoader.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/core/fs/factories/HadoopFileSystemFactoryLoader.java b/flink-core/src/main/java/org/apache/flink/core/fs/factories/HadoopFileSystemFactoryLoader.java
new file mode 100644
index 0000000..ed584ef
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/core/fs/factories/HadoopFileSystemFactoryLoader.java
@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.core.fs.factories;
+
+import org.apache.flink.core.fs.FileSystemFactory;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A utility class to check and reflectively load the Hadoop file system factory.
+ */
+public class HadoopFileSystemFactoryLoader {
+
+ private static final Logger LOG = LoggerFactory.getLogger(HadoopFileSystemFactoryLoader.class);
+
+ private static final String FACTORY_CLASS = "org.apache.flink.runtime.fs.hdfs.HadoopFsFactory";
+
+ private static final String HADOOP_CONFIG_CLASS = "org.apache.hadoop.conf.Configuration";
+
+ private static final String HADOOP_FS_CLASS = "org.apache.hadoop.fs.FileSystem";
+
+
+ /**
+ * Loads the FileSystemFactory for the Hadoop-backed file systems.
+ */
+ public static FileSystemFactory loadFactory() {
+ final ClassLoader cl = HadoopFileSystemFactoryLoader.class.getClassLoader();
+
+ // first, see if the Flink runtime classes are available
+ final Class<? extends FileSystemFactory> factoryClass;
+ try {
+ factoryClass = Class.forName(FACTORY_CLASS, false, cl).asSubclass(FileSystemFactory.class);
+ }
+ catch (ClassNotFoundException e) {
+ LOG.info("No Flink runtime dependency present - the extended set of supported File Systems " +
+ "via Hadoop is not available.");
+ return new UnsupportedSchemeFactory("Flink runtime classes missing in classpath/dependencies.");
+ }
+ catch (Exception | LinkageError e) {
+ LOG.warn("Flink's Hadoop file system factory could not be loaded", e);
+ return new UnsupportedSchemeFactory("Flink's Hadoop file system factory could not be loaded", e);
+ }
+
+ // check (for eager and better exception messages) if the Hadoop classes are available here
+ try {
+ Class.forName(HADOOP_CONFIG_CLASS, false, cl);
+ Class.forName(HADOOP_FS_CLASS, false, cl);
+ }
+ catch (ClassNotFoundException e) {
+ LOG.info("Hadoop is not in the classpath/dependencies - the extended set of supported File Systems " +
+ "via Hadoop is not available.");
+ return new UnsupportedSchemeFactory("Hadoop is not in the classpath/dependencies.");
+ }
+
+ // Create the factory.
+ try {
+ return factoryClass.newInstance();
+ }
+ catch (Exception | LinkageError e) {
+ LOG.warn("Flink's Hadoop file system factory could not be created", e);
+ return new UnsupportedSchemeFactory("Flink's Hadoop file system factory could not be created", e);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/536675b0/flink-core/src/main/java/org/apache/flink/core/fs/factories/LocalFileSystemFactory.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/core/fs/factories/LocalFileSystemFactory.java b/flink-core/src/main/java/org/apache/flink/core/fs/factories/LocalFileSystemFactory.java
new file mode 100644
index 0000000..fc04de5
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/core/fs/factories/LocalFileSystemFactory.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.core.fs.factories;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.FileSystemFactory;
+import org.apache.flink.core.fs.local.LocalFileSystem;
+
+import java.net.URI;
+
+/**
+ * A factory for the {@link LocalFileSystem}.
+ */
+@PublicEvolving
+public class LocalFileSystemFactory implements FileSystemFactory {
+
+ @Override
+ public void configure(Configuration config) {
+ // the local file system takes no configuration, so nothing to do here
+ }
+
+ @Override
+ public FileSystem create(URI fsUri) {
+ return LocalFileSystem.getSharedInstance();
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/536675b0/flink-core/src/main/java/org/apache/flink/core/fs/factories/MapRFsFactory.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/core/fs/factories/MapRFsFactory.java b/flink-core/src/main/java/org/apache/flink/core/fs/factories/MapRFsFactory.java
new file mode 100644
index 0000000..271e5bc
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/core/fs/factories/MapRFsFactory.java
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.core.fs.factories;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.FileSystemFactory;
+
+import java.io.IOException;
+import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
+import java.net.URI;
+
+/**
+ * A factory for the MapR file system.
+ *
+ * <p>This factory tries to reflectively instantiate the MapR file system. It can only be
+ * used when the MapR FS libraries are in the classpath.
+ */
+public class MapRFsFactory implements FileSystemFactory {
+
+ private static final String MAPR_FILESYSTEM_CLASS = "org.apache.flink.runtime.fs.maprfs.MapRFileSystem";
+
+ @Override
+ public void configure(Configuration config) {
+ // nothing to configure based on the configuration here
+ }
+
+ @Override
+ public FileSystem create(URI fsUri) throws IOException {
+ try {
+ Class<? extends FileSystem> fsClass = Class.forName(
+ MAPR_FILESYSTEM_CLASS, false, getClass().getClassLoader()).asSubclass(FileSystem.class);
+
+ Constructor<? extends FileSystem> constructor = fsClass.getConstructor(URI.class);
+
+ try {
+ return constructor.newInstance(fsUri);
+ }
+ catch (InvocationTargetException e) {
+ throw e.getTargetException();
+ }
+ }
+ catch (ClassNotFoundException e) {
+ throw new IOException("Could not load MapR file system class '" + MAPR_FILESYSTEM_CLASS +
+ "\'. Please make sure the Flink runtime classes are part of the classpath or dependencies.", e);
+ }
+ catch (LinkageError e) {
+ throw new IOException("Some of the MapR FS or required Hadoop classes seem to be missing or incompatible. "
+ + "Please check that a compatible version of the MapR Hadoop libraries is in the classpath.", e);
+ }
+ catch (IOException e) {
+ throw e;
+ }
+ catch (Throwable t) {
+ throw new IOException("Could not instantiate MapR file system class '" + MAPR_FILESYSTEM_CLASS + "'.", t);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/536675b0/flink-core/src/main/java/org/apache/flink/core/fs/factories/UnsupportedSchemeFactory.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/core/fs/factories/UnsupportedSchemeFactory.java b/flink-core/src/main/java/org/apache/flink/core/fs/factories/UnsupportedSchemeFactory.java
new file mode 100644
index 0000000..8464b63
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/core/fs/factories/UnsupportedSchemeFactory.java
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.core.fs.factories;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.FileSystemFactory;
+import org.apache.flink.core.fs.UnsupportedFileSystemSchemeException;
+
+import javax.annotation.Nullable;
+import java.io.IOException;
+import java.net.URI;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * A file system factory to throw an UnsupportedFileSystemSchemeException when called.
+ */
+public class UnsupportedSchemeFactory implements FileSystemFactory {
+
+ private final String exceptionMessage;
+
+ @Nullable
+ private final Throwable exceptionCause;
+
+ public UnsupportedSchemeFactory(String exceptionMessage) {
+ this(exceptionMessage, null);
+ }
+
+ public UnsupportedSchemeFactory(String exceptionMessage, @Nullable Throwable exceptionCause) {
+ this.exceptionMessage = checkNotNull(exceptionMessage);
+ this.exceptionCause = exceptionCause;
+ }
+
+ @Override
+ public void configure(Configuration config) {
+ // nothing to do here
+ }
+
+ @Override
+ public FileSystem create(URI fsUri) throws IOException {
+ if (exceptionCause == null) {
+ throw new UnsupportedFileSystemSchemeException(exceptionMessage);
+ }
+ else {
+ throw new UnsupportedFileSystemSchemeException(exceptionMessage, exceptionCause);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/536675b0/flink-core/src/main/java/org/apache/flink/core/fs/local/LocalFileSystem.java
----------------------------------------------------------------------
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 f21a481..ecfd21c 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
@@ -62,7 +62,10 @@ public class LocalFileSystem extends FileSystem {
private static final Logger LOG = LoggerFactory.getLogger(LocalFileSystem.class);
/** The URI representing the local file system. */
- private static final URI uri = OperatingSystem.isWindows() ? URI.create("file:/") : URI.create("file:///");
+ private static final URI LOCAL_URI = OperatingSystem.isWindows() ? URI.create("file:/") : URI.create("file:///");
+
+ /** The shared instance of the local file system */
+ private static final LocalFileSystem INSTANCE = new LocalFileSystem();
/** Path pointing to the current working directory.
* Because Paths are not immutable, we cannot cache the proper path here */
@@ -114,7 +117,7 @@ public class LocalFileSystem extends FileSystem {
@Override
public URI getUri() {
- return uri;
+ return LOCAL_URI;
}
@Override
@@ -128,9 +131,6 @@ public class LocalFileSystem extends FileSystem {
}
@Override
- public void initialize(final URI name) throws IOException {}
-
- @Override
public FSDataInputStream open(final Path f, final int bufferSize) throws IOException {
return open(f);
}
@@ -293,6 +293,15 @@ public class LocalFileSystem extends FileSystem {
* @return The URI that represents the local file system.
*/
public static URI getLocalFsURI() {
- return uri;
+ return LOCAL_URI;
+ }
+
+ /**
+ * Gets the shared instance of this file system.
+ *
+ * @return The shared instance of this file system.
+ */
+ public static LocalFileSystem getSharedInstance() {
+ return INSTANCE;
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/536675b0/flink-core/src/test/java/org/apache/flink/configuration/FilesystemSchemeConfigTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/configuration/FilesystemSchemeConfigTest.java b/flink-core/src/test/java/org/apache/flink/configuration/FilesystemSchemeConfigTest.java
index 0cf5e32..43c79c1 100644
--- a/flink-core/src/test/java/org/apache/flink/configuration/FilesystemSchemeConfigTest.java
+++ b/flink-core/src/test/java/org/apache/flink/configuration/FilesystemSchemeConfigTest.java
@@ -19,114 +19,86 @@
package org.apache.flink.configuration;
import org.apache.flink.core.fs.FileSystem;
-import org.apache.flink.core.fs.Path;
-import org.apache.flink.core.testutils.CommonTestUtils;
+import org.apache.flink.core.fs.UnsupportedFileSystemSchemeException;
+import org.apache.flink.core.fs.local.LocalFileSystem;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.After;
+import org.junit.Rule;
import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
-import java.io.File;
-import java.io.FileNotFoundException;
import java.io.IOException;
-import java.io.PrintWriter;
-import java.lang.reflect.Field;
import java.net.URI;
-import java.net.URISyntaxException;
-import java.util.UUID;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
-public class FilesystemSchemeConfigTest {
+public class FilesystemSchemeConfigTest extends TestLogger {
- @Test
- public void testExplicitFilesystemScheme() {
- testSettingFilesystemScheme(false, "fs.default-scheme: otherFS://localhost:1234/", true);
+ @Rule
+ public final TemporaryFolder tempFolder = new TemporaryFolder();
+
+ @After
+ public void clearFsSettings() throws IOException {
+ FileSystem.initialize(new Configuration());
}
+ // ------------------------------------------------------------------------
+
@Test
- public void testSettingFilesystemSchemeInConfiguration() {
- testSettingFilesystemScheme(false, "fs.default-scheme: file:///", false);
+ public void testDefaultsToLocal() throws Exception {
+ URI justPath = new URI(tempFolder.newFile().toURI().getPath());
+ assertNull(justPath.getScheme());
+
+ FileSystem fs = FileSystem.get(justPath);
+ assertEquals("file", fs.getUri().getScheme());
}
@Test
- public void testUsingDefaultFilesystemScheme() {
- testSettingFilesystemScheme(true, "fs.default-scheme: file:///", false);
+ public void testExplicitlySetToLocal() throws Exception {
+ final Configuration conf = new Configuration();
+ conf.setString(ConfigConstants.FILESYSTEM_SCHEME, LocalFileSystem.getLocalFsURI().toString());
+ FileSystem.initialize(conf);
+
+ URI justPath = new URI(tempFolder.newFile().toURI().getPath());
+ assertNull(justPath.getScheme());
+
+ FileSystem fs = FileSystem.get(justPath);
+ assertEquals("file", fs.getUri().getScheme());
}
- private void testSettingFilesystemScheme(boolean useDefaultScheme,
- String configFileScheme, boolean useExplicitScheme) {
- final File tmpDir = getTmpDir();
- final File confFile = new File(tmpDir, GlobalConfiguration.FLINK_CONF_FILENAME);
- try {
- confFile.createNewFile();
- } catch (IOException e) {
- throw new RuntimeException("Couldn't create file", e);
- }
- final File testFile = new File(tmpDir.getAbsolutePath() + File.separator + "testing.txt");
+ @Test
+ public void testExplicitlySetToOther() throws Exception {
+ final Configuration conf = new Configuration();
+ conf.setString(ConfigConstants.FILESYSTEM_SCHEME, "otherFS://localhost:1234/");
+ FileSystem.initialize(conf);
+
+ URI justPath = new URI(tempFolder.newFile().toURI().getPath());
+ assertNull(justPath.getScheme());
try {
- try {
- final PrintWriter pw1 = new PrintWriter(confFile);
- if(!useDefaultScheme) {
- pw1.println(configFileScheme);
- }
- pw1.close();
-
- final PrintWriter pwTest = new PrintWriter(testFile);
- pwTest.close();
-
- } catch (FileNotFoundException e) {
- fail(e.getMessage());
- }
-
- Configuration conf = GlobalConfiguration.loadConfiguration(tmpDir.getAbsolutePath());
-
- try {
- FileSystem.setDefaultScheme(conf);
- String noSchemePath = testFile.toURI().getPath(); // remove the scheme.
-
- URI uri = new URI(noSchemePath);
- // check if the scheme == null (so that we get the configuration one.
- assertTrue(uri.getScheme() == null);
-
- // get the filesystem with the default scheme as set in the confFile1
- FileSystem fs = useExplicitScheme ? FileSystem.get(testFile.toURI()) : FileSystem.get(uri);
- assertTrue(fs.exists(new Path(noSchemePath)));
-
- } catch (IOException e) {
- fail(e.getMessage());
- } catch (URISyntaxException e) {
- e.printStackTrace();
- }
- } finally {
- try {
- // clear the default scheme set in the FileSystem class.
- // we do it through reflection to avoid creating a publicly
- // accessible method, which could also be wrongly used by users.
-
- Field f = FileSystem.class.getDeclaredField("defaultScheme");
- f.setAccessible(true);
- f.set(null, null);
- } catch (IllegalAccessException | NoSuchFieldException e) {
- e.printStackTrace();
- fail("Cannot reset default scheme: " + e.getMessage());
- }
-
- confFile.delete();
- testFile.delete();
- tmpDir.delete();
+ FileSystem.get(justPath);
+ fail("should have failed with an exception");
+ }
+ catch (UnsupportedFileSystemSchemeException e) {
+ assertTrue(e.getMessage().contains("otherFS"));
}
}
- private File getTmpDir() {
- File tmpDir = new File(CommonTestUtils.getTempDir() + File.separator
- + UUID.randomUUID().toString() + File.separator);
-
- assertTrue(tmpDir.mkdirs());
+ @Test
+ public void testExplicitlyPathTakesPrecedence() throws Exception {
+ final Configuration conf = new Configuration();
+ conf.setString(ConfigConstants.FILESYSTEM_SCHEME, "otherFS://localhost:1234/");
+ FileSystem.initialize(conf);
- return tmpDir;
- }
+ URI pathAndScheme = tempFolder.newFile().toURI();
+ assertNotNull(pathAndScheme.getScheme());
- private File createRandomFile(File path, String suffix) {
- return new File(path.getAbsolutePath() + File.separator + UUID.randomUUID() + suffix);
+ FileSystem fs = FileSystem.get(pathAndScheme);
+ assertEquals("file", fs.getUri().getScheme());
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/536675b0/flink-core/src/test/java/org/apache/flink/testutils/TestFileSystem.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/testutils/TestFileSystem.java b/flink-core/src/test/java/org/apache/flink/testutils/TestFileSystem.java
index 00f88ea..1e5a7b0 100644
--- a/flink-core/src/test/java/org/apache/flink/testutils/TestFileSystem.java
+++ b/flink-core/src/test/java/org/apache/flink/testutils/TestFileSystem.java
@@ -23,9 +23,11 @@ import java.lang.reflect.Field;
import java.net.URI;
import java.util.Map;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.FSDataInputStream;
import org.apache.flink.core.fs.FileStatus;
import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.FileSystemFactory;
import org.apache.flink.core.fs.Path;
import org.apache.flink.core.fs.local.LocalFileStatus;
import org.apache.flink.core.fs.local.LocalFileSystem;
@@ -69,21 +71,34 @@ public class TestFileSystem extends LocalFileSystem {
}
return newStati;
}
-
+
+ @Override
+ public URI getUri() {
+ return URI.create("test:///");
+ }
+
public static void registerTestFileSysten() throws Exception {
Class<FileSystem> fsClass = FileSystem.class;
- Field dirField = fsClass.getDeclaredField("FSDIRECTORY");
-
+ Field dirField = fsClass.getDeclaredField("FS_FACTORIES");
+
dirField.setAccessible(true);
@SuppressWarnings("unchecked")
- Map<String, String> map = (Map<String, String>) dirField.get(null);
+ Map<String, FileSystemFactory> map = (Map<String, FileSystemFactory>) dirField.get(null);
dirField.setAccessible(false);
-
- map.put("test", TestFileSystem.class.getName());
+
+ map.put("test", new TestFileSystemFactory());
}
- @Override
- public URI getUri() {
- return URI.create("test:///");
+ // ------------------------------------------------------------------------
+
+ private static final class TestFileSystemFactory implements FileSystemFactory {
+
+ @Override
+ public void configure(Configuration config) {}
+
+ @Override
+ public FileSystem create(URI fsUri) throws IOException {
+ return new TestFileSystem();
+ }
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/536675b0/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosTaskExecutorRunner.java
----------------------------------------------------------------------
diff --git a/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosTaskExecutorRunner.java b/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosTaskExecutorRunner.java
index 37e5b31..41449be 100644
--- a/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosTaskExecutorRunner.java
+++ b/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosTaskExecutorRunner.java
@@ -99,12 +99,11 @@ public class MesosTaskExecutorRunner {
configuration.setString(ConfigConstants.TASK_MANAGER_TMP_DIR_KEY, tmpDirs);
}
- // configure the default filesystem
+ // configure the filesystems
try {
- FileSystem.setDefaultScheme(configuration);
+ FileSystem.initialize(configuration);
} catch (IOException e) {
- throw new IOException("Error while setting the default " +
- "filesystem scheme from configuration.", e);
+ throw new IOException("Error while configuring the filesystems.", e);
}
// tell akka to die in case of an error
http://git-wip-us.apache.org/repos/asf/flink/blob/536675b0/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosApplicationMasterRunner.java
----------------------------------------------------------------------
diff --git a/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosApplicationMasterRunner.java b/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosApplicationMasterRunner.java
index f465464..a6ea133 100755
--- a/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosApplicationMasterRunner.java
+++ b/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosApplicationMasterRunner.java
@@ -156,12 +156,11 @@ public class MesosApplicationMasterRunner {
final Configuration dynamicProperties = BootstrapTools.parseDynamicProperties(cmd);
final Configuration config = GlobalConfiguration.loadConfigurationWithDynamicProperties(dynamicProperties);
- // configure the default filesystem
+ // configure the filesystems
try {
- FileSystem.setDefaultScheme(config);
+ FileSystem.initialize(config);
} catch (IOException e) {
- throw new IOException("Error while setting the default " +
- "filesystem scheme from configuration.", e);
+ throw new IOException("Error while configuring the filesystems.", e);
}
// configure security
http://git-wip-us.apache.org/repos/asf/flink/blob/536675b0/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosTaskManagerRunner.java
----------------------------------------------------------------------
diff --git a/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosTaskManagerRunner.java b/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosTaskManagerRunner.java
index 67dca4a..4549970 100644
--- a/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosTaskManagerRunner.java
+++ b/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosTaskManagerRunner.java
@@ -99,12 +99,11 @@ public class MesosTaskManagerRunner {
configuration.setString(ConfigConstants.TASK_MANAGER_TMP_DIR_KEY, tmpDirs);
}
- // configure the default filesystem
+ // configure the filesystems
try {
- FileSystem.setDefaultScheme(configuration);
+ FileSystem.initialize(configuration);
} catch (IOException e) {
- throw new IOException("Error while setting the default " +
- "filesystem scheme from configuration.", e);
+ throw new IOException("Error while confoguring the filesystems.", e);
}
// tell akka to die in case of an error
http://git-wip-us.apache.org/repos/asf/flink/blob/536675b0/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java
index e9de7ad..b2ddf1d 100755
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java
@@ -103,7 +103,7 @@ public abstract class ClusterEntrypoint implements FatalErrorHandler {
LOG.info("Starting {}.", getClass().getSimpleName());
try {
- installDefaultFileSystem(configuration);
+ configureFileSystems(configuration);
SecurityContext securityContext = installSecurityContext(configuration);
@@ -128,11 +128,11 @@ public abstract class ClusterEntrypoint implements FatalErrorHandler {
}
}
- protected void installDefaultFileSystem(Configuration configuration) throws Exception {
+ protected void configureFileSystems(Configuration configuration) throws Exception {
LOG.info("Install default filesystem.");
try {
- FileSystem.setDefaultScheme(configuration);
+ FileSystem.initialize(configuration);
} catch (IOException e) {
throw new IOException("Error while setting the default " +
"filesystem scheme from configuration.", e);
http://git-wip-us.apache.org/repos/asf/flink/blob/536675b0/flink-runtime/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopFileSystem.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopFileSystem.java b/flink-runtime/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopFileSystem.java
index 7ab7ab7..fd6b9da 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopFileSystem.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopFileSystem.java
@@ -18,252 +18,47 @@
package org.apache.flink.runtime.fs.hdfs;
-import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.GlobalConfiguration;
import org.apache.flink.core.fs.BlockLocation;
import org.apache.flink.core.fs.FileStatus;
import org.apache.flink.core.fs.FileSystem;
-import org.apache.flink.core.fs.HadoopFileSystemWrapper;
import org.apache.flink.core.fs.Path;
-import org.apache.flink.util.InstantiationUtil;
+import org.apache.flink.runtime.util.HadoopUtils;
-import org.apache.hadoop.conf.Configuration;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.File;
import java.io.IOException;
-import java.lang.reflect.Method;
import java.net.URI;
-import java.net.UnknownHostException;
import static org.apache.flink.util.Preconditions.checkNotNull;
/**
- * Concrete implementation of the {@link FileSystem} base class for the Hadoop File System. The
- * class is a wrapper class which encapsulated the original Hadoop HDFS API.
- *
- * <p>If no file system class is specified, the wrapper will automatically load the Hadoop
- * distributed file system (HDFS).
- *
+ * A {@link FileSystem} that wraps an {@link org.apache.hadoop.fs.FileSystem Hadoop File System}.
*/
-public final class HadoopFileSystem extends FileSystem implements HadoopFileSystemWrapper {
-
- private static final Logger LOG = LoggerFactory.getLogger(HadoopFileSystem.class);
-
- private static final String DEFAULT_HDFS_CLASS = "org.apache.hadoop.hdfs.DistributedFileSystem";
-
- /**
- * Configuration value name for the DFS implementation name. Usually not specified in hadoop configurations.
- */
- private static final String HDFS_IMPLEMENTATION_KEY = "fs.hdfs.impl";
-
- private final org.apache.hadoop.conf.Configuration conf;
+public final class HadoopFileSystem extends FileSystem {
+ /** The wrapped Hadoop File System. */
private final org.apache.hadoop.fs.FileSystem fs;
-
/**
- * Creates a new DistributedFileSystem object to access HDFS, based on a class name
- * and picking up the configuration from the class path or the Flink configuration.
+ * Wraps the given Hadoop File System object as a Flink File System object.
+ * The given Hadoop file system object is expected to be initialized already.
*
- * @throws IOException
- * throw if the required HDFS classes cannot be instantiated
- */
- public HadoopFileSystem(Class<? extends org.apache.hadoop.fs.FileSystem> fsClass) throws IOException {
- // Create new Hadoop configuration object
- this.conf = getHadoopConfiguration();
-
- if (fsClass == null) {
- fsClass = getDefaultHDFSClass();
- }
-
- this.fs = instantiateFileSystem(fsClass);
- }
-
- /**
- * Creates a new DistributedFileSystem that uses the given Hadoop
- * {@link org.apache.hadoop.fs.FileSystem} under the hood.
- *
- * @param hadoopConfig The Hadoop configuration that the FileSystem is based on.
* @param hadoopFileSystem The Hadoop FileSystem that will be used under the hood.
*/
- public HadoopFileSystem(
- org.apache.hadoop.conf.Configuration hadoopConfig,
- org.apache.hadoop.fs.FileSystem hadoopFileSystem) {
-
- this.conf = checkNotNull(hadoopConfig, "hadoopConfig");
+ public HadoopFileSystem(org.apache.hadoop.fs.FileSystem hadoopFileSystem) {
this.fs = checkNotNull(hadoopFileSystem, "hadoopFileSystem");
}
- private Class<? extends org.apache.hadoop.fs.FileSystem> getDefaultHDFSClass() throws IOException {
- Class<? extends org.apache.hadoop.fs.FileSystem> fsClass = null;
-
- // try to get the FileSystem implementation class Hadoop 2.0.0 style
- {
- LOG.debug("Trying to load HDFS class Hadoop 2.x style.");
-
- Object fsHandle = null;
- try {
- Method newApi = org.apache.hadoop.fs.FileSystem.class.getMethod("getFileSystemClass", String.class, org.apache.hadoop.conf.Configuration.class);
- fsHandle = newApi.invoke(null, "hdfs", conf);
- } catch (Exception e) {
- // if we can't find the FileSystem class using the new API,
- // clazz will still be null, we assume we're running on an older Hadoop version
- }
-
- if (fsHandle != null) {
- if (fsHandle instanceof Class && org.apache.hadoop.fs.FileSystem.class.isAssignableFrom((Class<?>) fsHandle)) {
- fsClass = ((Class<?>) fsHandle).asSubclass(org.apache.hadoop.fs.FileSystem.class);
-
- if (LOG.isDebugEnabled()) {
- LOG.debug("Loaded '{}' as HDFS class.", fsClass.getName());
- }
- }
- else {
- LOG.debug("Unexpected return type from 'org.apache.hadoop.fs.FileSystem.getFileSystemClass(String, Configuration)'.");
- throw new RuntimeException("The value returned from org.apache.hadoop.fs.FileSystem.getFileSystemClass(String, Configuration) is not a valid subclass of org.apache.hadoop.fs.FileSystem.");
- }
- }
- }
-
- // fall back to an older Hadoop version
- if (fsClass == null) {
- // first of all, check for a user-defined hdfs class
- if (LOG.isDebugEnabled()) {
- LOG.debug("Falling back to loading HDFS class old Hadoop style. Looking for HDFS class configuration entry '{}'.",
- HDFS_IMPLEMENTATION_KEY);
- }
-
- Class<?> classFromConfig = conf.getClass(HDFS_IMPLEMENTATION_KEY, null);
-
- if (classFromConfig != null) {
- if (org.apache.hadoop.fs.FileSystem.class.isAssignableFrom(classFromConfig)) {
- fsClass = classFromConfig.asSubclass(org.apache.hadoop.fs.FileSystem.class);
-
- if (LOG.isDebugEnabled()) {
- LOG.debug("Loaded HDFS class '{}' as specified in configuration.", fsClass.getName());
- }
- }
- else {
- if (LOG.isDebugEnabled()) {
- LOG.debug("HDFS class specified by {} is of wrong type.", HDFS_IMPLEMENTATION_KEY);
- }
-
- throw new IOException("HDFS class specified by " + HDFS_IMPLEMENTATION_KEY +
- " cannot be cast to a FileSystem type.");
- }
- }
- else {
- // load the default HDFS class
- if (LOG.isDebugEnabled()) {
- LOG.debug("Trying to load default HDFS implementation {}.", DEFAULT_HDFS_CLASS);
- }
-
- try {
- Class <?> reflectedClass = Class.forName(DEFAULT_HDFS_CLASS);
- if (org.apache.hadoop.fs.FileSystem.class.isAssignableFrom(reflectedClass)) {
- fsClass = reflectedClass.asSubclass(org.apache.hadoop.fs.FileSystem.class);
- } else {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Default HDFS class is of wrong type.");
- }
-
- throw new IOException("The default HDFS class '" + DEFAULT_HDFS_CLASS +
- "' cannot be cast to a FileSystem type.");
- }
- }
- catch (ClassNotFoundException e) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Default HDFS class cannot be loaded.");
- }
-
- throw new IOException("No HDFS class has been configured and the default class '" +
- DEFAULT_HDFS_CLASS + "' cannot be loaded.");
- }
- }
- }
- return fsClass;
- }
-
/**
- * Returns a new Hadoop Configuration object using the path to the hadoop conf configured
- * in the main configuration (flink-conf.yaml).
- * This method is public because its being used in the HadoopDataSource.
+ * Gets the underlying Hadoop FileSystem.
+ * @return The underlying Hadoop FileSystem.
*/
- public static org.apache.hadoop.conf.Configuration getHadoopConfiguration() {
-
- org.apache.flink.configuration.Configuration flinkConfiguration =
- GlobalConfiguration.loadConfiguration();
-
- Configuration retConf = new org.apache.hadoop.conf.Configuration();
-
- // We need to load both core-site.xml and hdfs-site.xml to determine the default fs path and
- // the hdfs configuration
- // Try to load HDFS configuration from Hadoop's own configuration files
- // 1. approach: Flink configuration
- final String hdfsDefaultPath = flinkConfiguration.getString(ConfigConstants.HDFS_DEFAULT_CONFIG, null);
- if (hdfsDefaultPath != null) {
- retConf.addResource(new org.apache.hadoop.fs.Path(hdfsDefaultPath));
- } else {
- LOG.trace("{} configuration key for hdfs-default configuration file not set", ConfigConstants.HDFS_DEFAULT_CONFIG);
- }
-
- final String hdfsSitePath = flinkConfiguration.getString(ConfigConstants.HDFS_SITE_CONFIG, null);
- if (hdfsSitePath != null) {
- retConf.addResource(new org.apache.hadoop.fs.Path(hdfsSitePath));
- } else {
- LOG.trace("{} configuration key for hdfs-site configuration file not set", ConfigConstants.HDFS_SITE_CONFIG);
- }
-
- // 2. Approach environment variables
- String[] possibleHadoopConfPaths = new String[4];
- possibleHadoopConfPaths[0] = flinkConfiguration.getString(ConfigConstants.PATH_HADOOP_CONFIG, null);
- possibleHadoopConfPaths[1] = System.getenv("HADOOP_CONF_DIR");
-
- if (System.getenv("HADOOP_HOME") != null) {
- possibleHadoopConfPaths[2] = System.getenv("HADOOP_HOME") + "/conf";
- possibleHadoopConfPaths[3] = System.getenv("HADOOP_HOME") + "/etc/hadoop"; // hadoop 2.2
- }
-
- for (String possibleHadoopConfPath : possibleHadoopConfPaths) {
- if (possibleHadoopConfPath != null) {
- if (new File(possibleHadoopConfPath).exists()) {
- if (new File(possibleHadoopConfPath + "/core-site.xml").exists()) {
- retConf.addResource(new org.apache.hadoop.fs.Path(possibleHadoopConfPath + "/core-site.xml"));
- } else {
- LOG.debug("File {}/core-site.xml not found.", possibleHadoopConfPath);
- }
-
- if (new File(possibleHadoopConfPath + "/hdfs-site.xml").exists()) {
- retConf.addResource(new org.apache.hadoop.fs.Path(possibleHadoopConfPath + "/hdfs-site.xml"));
- } else {
- LOG.debug("File {}/hdfs-site.xml not found.", possibleHadoopConfPath);
- }
- }
- }
- }
- return retConf;
+ public org.apache.hadoop.fs.FileSystem getHadoopFileSystem() {
+ return this.fs;
}
- private org.apache.hadoop.fs.FileSystem instantiateFileSystem(Class<? extends org.apache.hadoop.fs.FileSystem> fsClass)
- throws IOException {
- try {
- return fsClass.newInstance();
- }
- catch (ExceptionInInitializerError e) {
- throw new IOException("The filesystem class '" + fsClass.getName() + "' throw an exception upon initialization.", e.getException());
- }
- catch (Throwable t) {
- String errorMessage = InstantiationUtil.checkForInstantiationError(fsClass);
- if (errorMessage != null) {
- throw new IOException("The filesystem class '" + fsClass.getName() + "' cannot be instantiated: " + errorMessage);
- } else {
- throw new IOException("An error occurred while instantiating the filesystem class '" +
- fsClass.getName() + "'.", t);
- }
- }
- }
+ // ------------------------------------------------------------------------
+ // file system methods
+ // ------------------------------------------------------------------------
@Override
public Path getWorkingDirectory() {
@@ -279,87 +74,6 @@ public final class HadoopFileSystem extends FileSystem implements HadoopFileSyst
return fs.getUri();
}
- /**
- * Gets the underlying Hadoop FileSystem.
- * @return The underlying Hadoop FileSystem.
- */
- public org.apache.hadoop.fs.FileSystem getHadoopFileSystem() {
- return this.fs;
- }
-
- @Override
- public void initialize(URI path) throws IOException {
-
- // If the authority is not part of the path, we initialize with the fs.defaultFS entry.
- if (path.getAuthority() == null) {
-
- String configEntry = this.conf.get("fs.defaultFS", null);
- if (configEntry == null) {
- // fs.default.name deprecated as of hadoop 2.2.0 http://hadoop.apache.org/docs/current/hadoop-project-dist/hadoop-common/DeprecatedProperties.html
- configEntry = this.conf.get("fs.default.name", null);
- }
-
- if (LOG.isDebugEnabled()) {
- LOG.debug("fs.defaultFS is set to {}", configEntry);
- }
-
- if (configEntry == null) {
- throw new IOException(getMissingAuthorityErrorPrefix(path) + "Either no default file system (hdfs) configuration was registered, " +
- "or that configuration did not contain an entry for the default file system (usually 'fs.defaultFS').");
- } else {
- try {
- URI initURI = URI.create(configEntry);
-
- if (initURI.getAuthority() == null) {
- throw new IOException(getMissingAuthorityErrorPrefix(path) + "Either no default file system was registered, " +
- "or the provided configuration contains no valid authority component (fs.default.name or fs.defaultFS) " +
- "describing the (hdfs namenode) host and port.");
- } else {
- try {
- this.fs.initialize(initURI, this.conf);
- }
- catch (IOException e) {
- throw new IOException(getMissingAuthorityErrorPrefix(path) +
- "Could not initialize the file system connection with the given default file system address: " + e.getMessage(), e);
- }
- }
- }
- catch (IllegalArgumentException e) {
- throw new IOException(getMissingAuthorityErrorPrefix(path) +
- "The configuration contains an invalid file system default name (fs.default.name or fs.defaultFS): " + configEntry);
- }
- }
- }
- else {
- // Initialize file system
- 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() + ") 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);
- }
- }
- }
-
- private static String getMissingAuthorityErrorPrefix(URI path) {
- return "The given HDFS file URI (" + path.toString() + ") did not describe the HDFS NameNode." +
- " The attempt to use a default HDFS configuration, as specified in the '" + ConfigConstants.HDFS_DEFAULT_CONFIG + "' or '" +
- ConfigConstants.HDFS_SITE_CONFIG + "' config parameter failed due to the following problem: ";
- }
-
@Override
public FileStatus getFileStatus(final Path f) throws IOException {
org.apache.hadoop.fs.FileStatus status = this.fs.getFileStatus(new org.apache.hadoop.fs.Path(f.toString()));
@@ -456,19 +170,26 @@ public final class HadoopFileSystem extends FileSystem implements HadoopFileSyst
return true;
}
- @Override
- public Class<?> getHadoopWrapperClassNameForFileSystem(String scheme) {
- Configuration hadoopConf = getHadoopConfiguration();
- Class<? extends org.apache.hadoop.fs.FileSystem> clazz = null;
- try {
- clazz = org.apache.hadoop.fs.FileSystem.getFileSystemClass(scheme, hadoopConf);
- } catch (IOException e) {
- LOG.info("Flink could not load the Hadoop File system implementation for scheme " + scheme);
- }
+ // ------------------------------------------------------------------------
+ // Miscellaneous Utilities
+ // ------------------------------------------------------------------------
- if (clazz != null && LOG.isDebugEnabled()) {
- LOG.debug("Flink supports {} with the Hadoop file system wrapper, impl {}", scheme, clazz);
- }
- return clazz;
+ /**
+ * Returns a new Hadoop Configuration object using the path to the hadoop conf configured
+ * in the main configuration (flink-conf.yaml).
+ * This method is public because its being used in the HadoopDataSource.
+ *
+ * @deprecated This method should not be used, because it dynamically (and possibly incorrectly)
+ * re-loads the Flink configuration.
+ * Use {@link HadoopUtils#getHadoopConfiguration(org.apache.flink.configuration.Configuration)}
+ * instead.
+ */
+ @Deprecated
+ public static org.apache.hadoop.conf.Configuration getHadoopConfiguration() {
+
+ org.apache.flink.configuration.Configuration flinkConfiguration =
+ GlobalConfiguration.loadConfiguration();
+
+ return HadoopUtils.getHadoopConfiguration(flinkConfiguration);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/536675b0/flink-runtime/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopFsFactory.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopFsFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopFsFactory.java
new file mode 100644
index 0000000..d3b1b89
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopFsFactory.java
@@ -0,0 +1,167 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.fs.hdfs;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.FileSystemFactory;
+import org.apache.flink.core.fs.UnsupportedFileSystemSchemeException;
+import org.apache.flink.runtime.util.HadoopUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.UnknownHostException;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * A file system factory for Hadoop-based file systems.
+ *
+ * <p>This factory calls Hadoop's mechanism to find a file system implementation for a given file
+ * system scheme (a {@link org.apache.hadoop.fs.FileSystem}) and wraps it as a Flink file system
+ * (a {@link org.apache.flink.core.fs.FileSystem}).
+ */
+public class HadoopFsFactory implements FileSystemFactory {
+
+ private static final Logger LOG = LoggerFactory.getLogger(HadoopFsFactory.class);
+
+ /** Hadoop's configuration for the file systems. */
+ private org.apache.hadoop.conf.Configuration hadoopConfig;
+
+ @Override
+ public void configure(Configuration config) {
+ hadoopConfig = HadoopUtils.getHadoopConfiguration(config);
+ }
+
+ @Override
+ public FileSystem create(URI fsUri) throws IOException {
+ checkNotNull(fsUri, "fsUri");
+
+ final String scheme = fsUri.getScheme();
+ checkArgument(scheme != null, "file system has null scheme");
+
+ // -- (1) get the loaded Hadoop config (or fall back to one loaded from the classpath)
+
+ final org.apache.hadoop.conf.Configuration hadoopConfig;
+ if (this.hadoopConfig != null) {
+ hadoopConfig = this.hadoopConfig;
+ }
+ else {
+ LOG.warn("Hadoop configuration has not been explicitly initialized prior to loading a Hadoop file system."
+ + " Using configuration from the classpath.");
+
+ hadoopConfig = new org.apache.hadoop.conf.Configuration();
+ }
+
+ // -- (2) create the proper URI to initialize the file system
+
+ final URI initUri;
+ if (fsUri.getAuthority() != null) {
+ initUri = fsUri;
+ }
+ else {
+ LOG.debug("URI {} does not specify file system authority, trying to load default authority (fs.defaultFS)");
+
+ String configEntry = hadoopConfig.get("fs.defaultFS", null);
+ if (configEntry == null) {
+ // fs.default.name deprecated as of hadoop 2.2.0 - see
+ // http://hadoop.apache.org/docs/current/hadoop-project-dist/hadoop-common/DeprecatedProperties.html
+ configEntry = hadoopConfig.get("fs.default.name", null);
+ }
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Hadoop's 'fs.defaultFS' is set to {}", configEntry);
+ }
+
+ if (configEntry == null) {
+ throw new IOException(getMissingAuthorityErrorPrefix(fsUri) +
+ "Hadoop configuration did not contain an entry for the default file system ('fs.defaultFS').");
+ }
+ else {
+ try {
+ initUri = URI.create(configEntry);
+ }
+ catch (IllegalArgumentException e) {
+ throw new IOException(getMissingAuthorityErrorPrefix(fsUri) +
+ "The configuration contains an invalid file system default name " +
+ "('fs.default.name' or 'fs.defaultFS'): " + configEntry);
+ }
+
+ if (initUri.getAuthority() == null) {
+ throw new IOException(getMissingAuthorityErrorPrefix(fsUri) +
+ "Hadoop configuration for default file system ('fs.default.name' or 'fs.defaultFS') " +
+ "contains no valid authority component (like hdfs namenode, S3 host, etc)");
+ }
+ }
+ }
+
+ // -- (3) get the Hadoop file system class for that scheme
+
+ final Class<? extends org.apache.hadoop.fs.FileSystem> fsClass;
+ try {
+ fsClass = org.apache.hadoop.fs.FileSystem.getFileSystemClass(scheme, hadoopConfig);
+ }
+ catch (IOException e) {
+ throw new UnsupportedFileSystemSchemeException(
+ "Hadoop File System abstraction does not support scheme '" + scheme + "'. " +
+ "Either no file system implementation exists for that scheme, " +
+ "or the relevant classes are missing from the classpath.", e);
+ }
+
+ LOG.debug("Instantiating for file system scheme {} Hadoop File System {}", scheme, fsClass.getName());
+
+ // -- (4) instantiate the Hadoop file system
+
+ final org.apache.hadoop.fs.FileSystem hadoopFs;
+ try {
+ hadoopFs = fsClass.newInstance();
+ }
+ catch (Exception e) {
+ throw new IOException("The Hadoop file system class '" + fsClass.getName() + "' cannot be instantiated.", e);
+ }
+
+ // -- (5) configure the Hadoop file system
+
+ try {
+ hadoopFs.initialize(initUri, hadoopConfig);
+ }
+ catch (UnknownHostException e) {
+ String message = "The Hadoop file system's authority (" + initUri.getAuthority() +
+ "), specified by either the file URI or the configuration, cannot be resolved.";
+
+ throw new IOException(message, e);
+ }
+ catch (Exception e) {
+ throw new IOException("Hadoop file system " + fsClass.getName() + " for scheme '" + scheme +
+ "' could not ne initialized.", e);
+ }
+
+ return new HadoopFileSystem(hadoopFs);
+ }
+
+ private static String getMissingAuthorityErrorPrefix(URI fsURI) {
+ return "The given file system URI (" + fsURI.toString() + ") did not describe the authority " +
+ "(like for example HDFS NameNode address/port or S3 host). " +
+ "The attempt to use a configured default authority failed: ";
+ }
+}
[2/4] flink git commit: [FLINK-7643] [core] Misc. cleanups in
FileSystem
Posted by se...@apache.org.
[FLINK-7643] [core] Misc. cleanups in FileSystem
- Simplify access to local file system
- Use a fair lock for all FileSystem.get() operations
- Robust falback to local fs for default scheme (avoids URI parsing error on Windows)
- Deprecate 'getDefaultBlockSize()'
- Deprecate create(...) with block sizes and replication factor, which is not applicable to many FS
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/3b786844
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/3b786844
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/3b786844
Branch: refs/heads/master
Commit: 3b786844dd9c0ce176eac98c8a05ebe50cb1ebe7
Parents: 84a07a3
Author: Stephan Ewen <se...@apache.org>
Authored: Mon Oct 2 14:34:27 2017 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Thu Oct 5 19:14:45 2017 +0200
----------------------------------------------------------------------
.../org/apache/flink/core/fs/FileSystem.java | 133 +++++++++++--------
.../flink/core/fs/local/LocalFileSystem.java | 20 ++-
2 files changed, 93 insertions(+), 60 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/3b786844/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 fab0f4d..7c69425 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
@@ -30,7 +30,6 @@ import org.apache.flink.annotation.Public;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.local.LocalFileSystem;
-import org.apache.flink.util.OperatingSystem;
import javax.annotation.Nullable;
import java.io.File;
@@ -195,6 +194,8 @@ public abstract class FileSystem {
}
// ------------------------------------------------------------------------
+ // File System Implementation Classes
+ // ------------------------------------------------------------------------
private static final String HADOOP_WRAPPER_FILESYSTEM_CLASS = "org.apache.flink.runtime.fs.hdfs.HadoopFileSystem";
@@ -202,24 +203,33 @@ public abstract class FileSystem {
private static final String HADOOP_WRAPPER_SCHEME = "hdwrapper";
+ // ------------------------------------------------------------------------
+
/** This lock guards the methods {@link #initOutPathLocalFS(Path, WriteMode, boolean)} and
* {@link #initOutPathDistFS(Path, WriteMode, boolean)} which are otherwise susceptible to races */
private static final ReentrantLock OUTPUT_DIRECTORY_INIT_LOCK = new ReentrantLock(true);
- // ------------------------------------------------------------------------
-
/** Object used to protect calls to specific methods.*/
- private static final Object SYNCHRONIZATION_OBJECT = new Object();
+ private static final ReentrantLock LOCK = new ReentrantLock(true);
- /**
- * Data structure mapping file system keys (scheme + authority) to cached file system objects.
- */
- private static final Map<FSKey, FileSystem> CACHE = new HashMap<FSKey, FileSystem>();
+ /** Cache for file systems, by scheme + authority. */
+ private static final Map<FSKey, FileSystem> CACHE = new HashMap<>();
- /**
- * Data structure mapping file system schemes to the corresponding implementations
- */
- private static final Map<String, String> FSDIRECTORY = new HashMap<String, String>();
+ /** Mapping of file system schemes to the corresponding implementations */
+ private static final Map<String, String> FSDIRECTORY = new HashMap<>();
+
+ /** The local file system. Needs to be lazily initialized to avoid that some JVMs deadlock
+ * on static subclass initialization. */
+ private static LocalFileSystem LOCAL_FS;
+
+ /** The default filesystem scheme to be used, configured during process-wide initialization.
+ * This value defaults to the local file systems scheme {@code 'file:///'} or
+ * {@code 'file:/'}. */
+ private static URI defaultScheme;
+
+ // ------------------------------------------------------------------------
+ // Initialization
+ // ------------------------------------------------------------------------
static {
FSDIRECTORY.put("hdfs", HADOOP_WRAPPER_FILESYSTEM_CLASS);
@@ -228,32 +238,6 @@ public abstract class FileSystem {
}
/**
- * Returns a reference to the {@link FileSystem} instance for accessing the
- * local file system.
- *
- * @return a reference to the {@link FileSystem} instance for accessing the
- * local file system.
- */
- public static FileSystem getLocalFileSystem() {
- // this should really never fail.
- try {
- URI localUri = OperatingSystem.isWindows() ? new URI("file:/") : new URI("file:///");
- return get(localUri);
- }
- catch (Exception e) {
- throw new RuntimeException("Cannot create URI for local file system");
- }
- }
-
- /**
- * The default filesystem scheme to be used. This can be specified by the parameter
- * <code>fs.default-scheme</code> in <code>flink-conf.yaml</code>. By default this is
- * set to <code>file:///</code> (see {@link ConfigConstants#FILESYSTEM_SCHEME}
- * and {@link ConfigConstants#DEFAULT_FILESYSTEM_SCHEME}), and uses the local filesystem.
- */
- private static URI defaultScheme;
-
- /**
* <p>
* Sets the default filesystem scheme based on the user-specified configuration parameter
* <code>fs.default-scheme</code>. By default this is set to <code>file:///</code>
@@ -269,26 +253,55 @@ public abstract class FileSystem {
* @param config the configuration from where to fetch the parameter.
*/
public static void setDefaultScheme(Configuration config) throws IOException {
- synchronized (SYNCHRONIZATION_OBJECT) {
+ LOCK.lock();
+ try {
if (defaultScheme == null) {
- String stringifiedUri = config.getString(ConfigConstants.FILESYSTEM_SCHEME,
- ConfigConstants.DEFAULT_FILESYSTEM_SCHEME);
- try {
- defaultScheme = new URI(stringifiedUri);
- } catch (URISyntaxException e) {
- throw new IOException("The URI used to set the default filesystem " +
- "scheme ('" + stringifiedUri + "') is not valid.");
+ final String stringifiedUri = config.getString(ConfigConstants.FILESYSTEM_SCHEME, null);
+ if (stringifiedUri == null) {
+ defaultScheme = LocalFileSystem.getLocalFsURI();
+ }
+ else {
+ try {
+ defaultScheme = new URI(stringifiedUri);
+ } catch (URISyntaxException e) {
+ throw new IOException("The URI used to set the default filesystem " +
+ "scheme ('" + stringifiedUri + "') is not valid.");
+ }
}
}
}
+ finally {
+ LOCK.unlock();
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // Obtaining File System Instances
+ // ------------------------------------------------------------------------
+
+ /**
+ * Returns a reference to the {@link FileSystem} instance for accessing the local file system.
+ *
+ * @return a reference to the {@link FileSystem} instance for accessing the local file system.
+ */
+ public static FileSystem getLocalFileSystem() {
+ LOCK.lock();
+ try {
+ if (LOCAL_FS == null) {
+ LOCAL_FS = new LocalFileSystem();
+ }
+ return FileSystemSafetyNet.wrapWithSafetyNetWhenActivated(LOCAL_FS);
+ } finally {
+ LOCK.unlock();
+ }
}
@Internal
public static FileSystem getUnguardedFileSystem(URI uri) throws IOException {
- FileSystem fs;
+ final URI asked = uri;
- URI asked = uri;
- synchronized (SYNCHRONIZATION_OBJECT) {
+ LOCK.lock();
+ try {
if (uri.getScheme() == null) {
try {
@@ -333,6 +346,7 @@ public abstract class FileSystem {
}
// Try to create a new file system
+ final FileSystem fs;
if (!isFlinkSupportedScheme(uri.getScheme())) {
// no build in support for this file system. Falling back to Hadoop's FileSystem impl.
@@ -369,9 +383,12 @@ public abstract class FileSystem {
// Add new file system object to cache
CACHE.put(key, fs);
}
- }
- return fs;
+ return fs;
+ }
+ finally {
+ LOCK.unlock();
+ }
}
/**
@@ -515,7 +532,10 @@ public abstract class FileSystem {
* Return the number of bytes that large input files should be optimally be split into to minimize I/O time.
*
* @return the number of bytes that large input files should be optimally be split into to minimize I/O time
+ *
+ * @deprecated This value is no longer used and is meaningless.
*/
+ @Deprecated
public long getDefaultBlockSize() {
return 32 * 1024 * 1024; // 32 MB;
}
@@ -598,8 +618,15 @@ public abstract class FileSystem {
* Control the behavior of specific file systems via configurations instead.
*/
@Deprecated
- public abstract FSDataOutputStream create(Path f, boolean overwrite, int bufferSize, short replication,
- long blockSize) throws IOException;
+ public FSDataOutputStream create(
+ Path f,
+ boolean overwrite,
+ int bufferSize,
+ short replication,
+ long blockSize) throws IOException {
+
+ return create(f, overwrite ? WriteMode.OVERWRITE : WriteMode.NO_OVERWRITE);
+ }
/**
* Opens an FSDataOutputStream at the indicated Path.
http://git-wip-us.apache.org/repos/asf/flink/blob/3b786844/flink-core/src/main/java/org/apache/flink/core/fs/local/LocalFileSystem.java
----------------------------------------------------------------------
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 0e3e9f3..f21a481 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
@@ -258,13 +258,6 @@ public class LocalFileSystem extends FileSystem {
}
@Override
- public FSDataOutputStream create(
- Path f, boolean overwrite, int bufferSize, short replication, long blockSize) throws IOException {
- return create(f, overwrite ? WriteMode.OVERWRITE : WriteMode.NO_OVERWRITE);
- }
-
-
- @Override
public boolean rename(final Path src, final Path dst) throws IOException {
final File srcFile = pathToFile(src);
final File dstFile = pathToFile(dst);
@@ -289,4 +282,17 @@ public class LocalFileSystem extends FileSystem {
public boolean isDistributedFS() {
return false;
}
+
+ // ------------------------------------------------------------------------
+
+ /**
+ * Gets the URI that represents the local file system.
+ * That URI is {@code "file:/"} on Windows platforms and {@code "file:///"} on other
+ * UNIX family platforms.
+ *
+ * @return The URI that represents the local file system.
+ */
+ public static URI getLocalFsURI() {
+ return uri;
+ }
}