private static FileSystemFactory loadHadoopFsFactory ( ) { final ClassLoader cl = FileSystem . class . getClassLoader ( ) ; // first, see if the Flink runtime classes are available final Class < ? extends FileSystemFactory > factoryClass ; try { factoryClass = Class . forName ( "org.apache.flink.runtime.fs.hdfs.HadoopFsFactory" , 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 ( "org.apache.hadoop.conf.Configuration" , false , cl ) ; Class . forName ( "org.apache.hadoop.fs.FileSystem" , 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 ) ; } }
Utility loader for the Hadoop file system factory . We treat the Hadoop FS factory in a special way because we use it as a catch all for file systems schemes not supported directly in Flink .