@Test public void hadoopShouldLoadFileSystemWithMultipleZkUri() throws Exception { org.apache.hadoop.conf.Configuration conf = getConf(); URI uri = URI.create(Constants.HEADER + "zk@host1:2181,host2:2181,host3:2181/tmp/path.txt"); org.apache.hadoop.fs.FileSystem fs = org.apache.hadoop.fs.FileSystem.get(uri, conf); FileSystem hfs = getHadoopFilesystem(fs); assertTrue(hfs.mFileSystem.getConf().getBoolean(PropertyKey.ZOOKEEPER_ENABLED)); assertEquals("host1:2181,host2:2181,host3:2181", hfs.mFileSystem.getConf().get(PropertyKey.ZOOKEEPER_ADDRESS)); uri = URI.create(Constants.HEADER + "zk@host1:2181;host2:2181;host3:2181/tmp/path.txt"); fs = getHadoopFilesystem(org.apache.hadoop.fs.FileSystem.get(uri, conf)); assertTrue(hfs.mFileSystem.getConf().getBoolean(PropertyKey.ZOOKEEPER_ENABLED)); assertEquals("host1:2181,host2:2181,host3:2181", hfs.mFileSystem.getConf().get(PropertyKey.ZOOKEEPER_ADDRESS)); }
@Override public FSDataOutputStream create(Path path, FsPermission permission, boolean overwrite, int bufferSize, short replication, long blockSize, Progressable progress) throws IOException { LOG.debug("create({}, {}, {}, {}, {}, {}, {})", path, permission, overwrite, bufferSize, replication, blockSize, progress); if (mStatistics != null) { mStatistics.incrementWriteOps(1); } AlluxioURI uri = getAlluxioPath(path); CreateFilePOptions options = CreateFilePOptions.newBuilder().setBlockSizeBytes(blockSize) .setMode(new Mode(permission.toShort()).toProto()).setRecursive(true).build(); FileOutStream outStream; try { outStream = mFileSystem.createFile(uri, options); } catch (AlluxioException e) { try { if (mFileSystem.exists(uri)) { if (!overwrite) { throw new IOException( "Not allowed to create() (overwrite=false) for existing Alluxio path: " + uri); } if (mFileSystem.getStatus(uri).isFolder()) { throw new IOException( ExceptionMessage.FILE_CREATE_IS_DIRECTORY.getMessage(uri)); } mFileSystem.delete(uri); } outStream = mFileSystem.createFile(uri, options); } catch (AlluxioException e2) { throw new IOException(e2); } } return new FSDataOutputStream(outStream, mStatistics); }
AbstractFileSystem extends org.apache.hadoop.fs.FileSystem { @Override public FSDataOutputStream create(Path path, FsPermission permission, boolean overwrite, int bufferSize, short replication, long blockSize, Progressable progress) throws IOException { LOG.debug("create({}, {}, {}, {}, {}, {}, {})", path, permission, overwrite, bufferSize, replication, blockSize, progress); if (mStatistics != null) { mStatistics.incrementWriteOps(1); } AlluxioURI uri = getAlluxioPath(path); CreateFilePOptions options = CreateFilePOptions.newBuilder().setBlockSizeBytes(blockSize) .setMode(new Mode(permission.toShort()).toProto()).setRecursive(true).build(); FileOutStream outStream; try { outStream = mFileSystem.createFile(uri, options); } catch (AlluxioException e) { try { if (mFileSystem.exists(uri)) { if (!overwrite) { throw new IOException( "Not allowed to create() (overwrite=false) for existing Alluxio path: " + uri); } if (mFileSystem.getStatus(uri).isFolder()) { throw new IOException( ExceptionMessage.FILE_CREATE_IS_DIRECTORY.getMessage(uri)); } mFileSystem.delete(uri); } outStream = mFileSystem.createFile(uri, options); } catch (AlluxioException e2) { throw new IOException(e2); } } return new FSDataOutputStream(outStream, mStatistics); } }
AbstractFileSystem extends org.apache.hadoop.fs.FileSystem { @Override public FSDataOutputStream create(Path path, FsPermission permission, boolean overwrite, int bufferSize, short replication, long blockSize, Progressable progress) throws IOException { LOG.debug("create({}, {}, {}, {}, {}, {}, {})", path, permission, overwrite, bufferSize, replication, blockSize, progress); if (mStatistics != null) { mStatistics.incrementWriteOps(1); } AlluxioURI uri = getAlluxioPath(path); CreateFilePOptions options = CreateFilePOptions.newBuilder().setBlockSizeBytes(blockSize) .setMode(new Mode(permission.toShort()).toProto()).setRecursive(true).build(); FileOutStream outStream; try { outStream = mFileSystem.createFile(uri, options); } catch (AlluxioException e) { try { if (mFileSystem.exists(uri)) { if (!overwrite) { throw new IOException( "Not allowed to create() (overwrite=false) for existing Alluxio path: " + uri); } if (mFileSystem.getStatus(uri).isFolder()) { throw new IOException( ExceptionMessage.FILE_CREATE_IS_DIRECTORY.getMessage(uri)); } mFileSystem.delete(uri); } outStream = mFileSystem.createFile(uri, options); } catch (AlluxioException e2) { throw new IOException(e2); } } return new FSDataOutputStream(outStream, mStatistics); } protected  AbstractFileSystem(FileSystem fileSystem); protected  AbstractFileSystem(); }
AbstractFileSystem extends org.apache.hadoop.fs.FileSystem { @Override public FSDataOutputStream create(Path path, FsPermission permission, boolean overwrite, int bufferSize, short replication, long blockSize, Progressable progress) throws IOException { LOG.debug("create({}, {}, {}, {}, {}, {}, {})", path, permission, overwrite, bufferSize, replication, blockSize, progress); if (mStatistics != null) { mStatistics.incrementWriteOps(1); } AlluxioURI uri = getAlluxioPath(path); CreateFilePOptions options = CreateFilePOptions.newBuilder().setBlockSizeBytes(blockSize) .setMode(new Mode(permission.toShort()).toProto()).setRecursive(true).build(); FileOutStream outStream; try { outStream = mFileSystem.createFile(uri, options); } catch (AlluxioException e) { try { if (mFileSystem.exists(uri)) { if (!overwrite) { throw new IOException( "Not allowed to create() (overwrite=false) for existing Alluxio path: " + uri); } if (mFileSystem.getStatus(uri).isFolder()) { throw new IOException( ExceptionMessage.FILE_CREATE_IS_DIRECTORY.getMessage(uri)); } mFileSystem.delete(uri); } outStream = mFileSystem.createFile(uri, options); } catch (AlluxioException e2) { throw new IOException(e2); } } return new FSDataOutputStream(outStream, mStatistics); } protected  AbstractFileSystem(FileSystem fileSystem); protected  AbstractFileSystem(); @Override FSDataOutputStream append(Path path, int bufferSize, Progressable progress); @Override void close(); @Override FSDataOutputStream create(Path path, FsPermission permission, boolean overwrite,
      int bufferSize, short replication, long blockSize, Progressable progress); @Override @Deprecated FSDataOutputStream createNonRecursive(Path path, FsPermission permission,
      boolean overwrite, int bufferSize, short replication, long blockSize, Progressable progress); @Override @Deprecated boolean delete(Path path); @Override boolean delete(Path path, boolean recursive); @Override long getDefaultBlockSize(); @Nullable @Override BlockLocation[] getFileBlockLocations(FileStatus file, long start, long len); @Override short getDefaultReplication(); @Override boolean setReplication(Path path, short replication); @Override FileStatus getFileStatus(Path path); @Override void setOwner(Path path, final String username, final String groupname); @Override void setPermission(Path path, FsPermission permission); abstract String getScheme(); @Override URI getUri(); @Override Path getWorkingDirectory(); @Override synchronized void initialize(URI uri, org.apache.hadoop.conf.Configuration conf); @VisibleForTesting synchronized void initialize(URI uri, org.apache.hadoop.conf.Configuration conf,
      @Nullable AlluxioConfiguration alluxioConfiguration); @Override FileStatus[] listStatus(Path path); @Override boolean mkdirs(Path path, FsPermission permission); @Override FSDataInputStream open(Path path, int bufferSize); @Override boolean rename(Path src, Path dst); @Override void setWorkingDirectory(Path path); }
AbstractFileSystem extends org.apache.hadoop.fs.FileSystem { @Override public FSDataOutputStream create(Path path, FsPermission permission, boolean overwrite, int bufferSize, short replication, long blockSize, Progressable progress) throws IOException { LOG.debug("create({}, {}, {}, {}, {}, {}, {})", path, permission, overwrite, bufferSize, replication, blockSize, progress); if (mStatistics != null) { mStatistics.incrementWriteOps(1); } AlluxioURI uri = getAlluxioPath(path); CreateFilePOptions options = CreateFilePOptions.newBuilder().setBlockSizeBytes(blockSize) .setMode(new Mode(permission.toShort()).toProto()).setRecursive(true).build(); FileOutStream outStream; try { outStream = mFileSystem.createFile(uri, options); } catch (AlluxioException e) { try { if (mFileSystem.exists(uri)) { if (!overwrite) { throw new IOException( "Not allowed to create() (overwrite=false) for existing Alluxio path: " + uri); } if (mFileSystem.getStatus(uri).isFolder()) { throw new IOException( ExceptionMessage.FILE_CREATE_IS_DIRECTORY.getMessage(uri)); } mFileSystem.delete(uri); } outStream = mFileSystem.createFile(uri, options); } catch (AlluxioException e2) { throw new IOException(e2); } } return new FSDataOutputStream(outStream, mStatistics); } protected  AbstractFileSystem(FileSystem fileSystem); protected  AbstractFileSystem(); @Override FSDataOutputStream append(Path path, int bufferSize, Progressable progress); @Override void close(); @Override FSDataOutputStream create(Path path, FsPermission permission, boolean overwrite,
      int bufferSize, short replication, long blockSize, Progressable progress); @Override @Deprecated FSDataOutputStream createNonRecursive(Path path, FsPermission permission,
      boolean overwrite, int bufferSize, short replication, long blockSize, Progressable progress); @Override @Deprecated boolean delete(Path path); @Override boolean delete(Path path, boolean recursive); @Override long getDefaultBlockSize(); @Nullable @Override BlockLocation[] getFileBlockLocations(FileStatus file, long start, long len); @Override short getDefaultReplication(); @Override boolean setReplication(Path path, short replication); @Override FileStatus getFileStatus(Path path); @Override void setOwner(Path path, final String username, final String groupname); @Override void setPermission(Path path, FsPermission permission); abstract String getScheme(); @Override URI getUri(); @Override Path getWorkingDirectory(); @Override synchronized void initialize(URI uri, org.apache.hadoop.conf.Configuration conf); @VisibleForTesting synchronized void initialize(URI uri, org.apache.hadoop.conf.Configuration conf,
      @Nullable AlluxioConfiguration alluxioConfiguration); @Override FileStatus[] listStatus(Path path); @Override boolean mkdirs(Path path, FsPermission permission); @Override FSDataInputStream open(Path path, int bufferSize); @Override boolean rename(Path src, Path dst); @Override void setWorkingDirectory(Path path); static final String FIRST_COM_PATH; }