docs: public static FileSystem createHadoopFileSystem(
  			Path path,
  			@Nullable Configuration extraUserConf) throws IOException {
  
  		// try to get the Hadoop File System via the Flink File Systems
  		// that way we get the proper configuration
  
  		final org.apache.flink.core.fs.FileSystem flinkFs =
  				org.apache.flink.core.fs.FileSystem.getUnguardedFileSystem(path.toUri());
  		final FileSystem hadoopFs = (flinkFs instanceof HadoopFileSystem) ?
  				((HadoopFileSystem) flinkFs).getHadoopFileSystem() : null;
  
  		// fast path: if the Flink file system wraps Hadoop anyways and we need no extra config,
  		// then we use it directly
  		if (extraUserConf == null && hadoopFs != null) {
  			return hadoopFs;
  		}
  		else {
  			// we need to re-instantiate the Hadoop file system, because we either have
  			// a special config, or the Path gave us a Flink FS that is not backed by
  			// Hadoop (like file://)
  
  			final org.apache.hadoop.conf.Configuration hadoopConf;
  			if (hadoopFs != null) {
  				// have a Hadoop FS but need to apply extra config
  				hadoopConf = hadoopFs.getConf();
  			}
  			else {
  				// the Path gave us a Flink FS that is not backed by Hadoop (like file://)
  				// we need to get access to the Hadoop file system first
  
  				// we access the Hadoop FS in Flink, which carries the proper
  				// Hadoop configuration. we should get rid of this once the bucketing sink is
  				// properly implemented against Flink's FS abstraction
  
  				URI genericHdfsUri = URI.create("hdfs://localhost:12345/");
  				org.apache.flink.core.fs.FileSystem accessor =
  						org.apache.flink.core.fs.FileSystem.getUnguardedFileSystem(genericHdfsUri);
  
  				if (!(accessor instanceof HadoopFileSystem)) {
  					throw new IOException(
  							"Cannot instantiate a Hadoop file system to access the Hadoop configuration. " +
  							"FS for hdfs:// is " + accessor.getClass().getName());
  				}
  
  				hadoopConf = ((HadoopFileSystem) accessor).getHadoopFileSystem().getConf();
  			}
  
  			// finalize the configuration
  
  			final org.apache.hadoop.conf.Configuration finalConf;
  			if (extraUserConf == null) {
  				finalConf = hadoopConf;
  			}
  			else {
  				finalConf = new org.apache.hadoop.conf.Configuration(hadoopConf);
  
  				for (String key : extraUserConf.keySet()) {
  					finalConf.set(key, extraUserConf.getString(key, null));
  				}
  			}
  
  			// we explicitly re-instantiate the file system here in order to make sure
  			// that the configuration is applied.
  
  			URI fsUri = path.toUri();
  			final String scheme = fsUri.getScheme();
  			final String authority = fsUri.getAuthority();
  
  			if (scheme == null && authority == null) {
  				fsUri = FileSystem.getDefaultUri(finalConf);
  			}
  			else if (scheme != null && authority == null) {
  				URI defaultUri = FileSystem.getDefaultUri(finalConf);
  				if (scheme.equals(defaultUri.getScheme()) && defaultUri.getAuthority() != null) {
  					fsUri = defaultUri;
  				}
  			}
  
  			final Class<? extends FileSystem> fsClass = FileSystem.getFileSystemClass(fsUri.getScheme(), finalConf);
  			final FileSystem fs;
  			try {
  				fs = fsClass.newInstance();
  			}
  			catch (Exception e) {
  				throw new IOException("Cannot instantiate the Hadoop file system", e);
  			}
  
  			fs.initialize(fsUri, finalConf);
  
  			// We don't perform checksums on Hadoop's local filesystem and use the raw filesystem.
  			// Otherwise buffers are not flushed entirely during checkpointing which results in data loss.
  			if (fs instanceof LocalFileSystem) {
  				return ((LocalFileSystem) fs).getRaw();
  			}
  			return fs;
  		}
  	}