FTPFileSystem extends FileSystem { @VisibleForTesting int getTransferMode(Configuration conf) { final String mode = conf.get(FS_FTP_TRANSFER_MODE); int ret = FTP.BLOCK_TRANSFER_MODE; if (mode == null) { return ret; } final String upper = mode.toUpperCase(); if (upper.equals("STREAM_TRANSFER_MODE")) { ret = FTP.STREAM_TRANSFER_MODE; } else if (upper.equals("COMPRESSED_TRANSFER_MODE")) { ret = FTP.COMPRESSED_TRANSFER_MODE; } else { if (!upper.equals("BLOCK_TRANSFER_MODE")) { LOG.warn("Cannot parse the value for " + FS_FTP_TRANSFER_MODE + ": " + mode + ". Using default."); } } return ret; }  @Override String getScheme(); @Override void initialize(URI uri, Configuration conf); @Override FSDataInputStream open(Path file, int bufferSize); @Override FSDataOutputStream create(Path file, FsPermission permission,
      boolean overwrite, int bufferSize, short replication, long blockSize,
      Progressable progress); @Override FSDataOutputStream append(Path f, int bufferSize,
      Progressable progress); @Override boolean delete(Path file, boolean recursive); @Override URI getUri(); @Override FileStatus[] listStatus(Path file); @Override FileStatus getFileStatus(Path file); @Override boolean mkdirs(Path file, FsPermission permission); @Override boolean rename(Path src, Path dst); @Override Path getWorkingDirectory(); @Override Path getHomeDirectory(); @Override void setWorkingDirectory(Path newDir); static final Logger LOG; static final int DEFAULT_BUFFER_SIZE; static final int DEFAULT_BLOCK_SIZE; static final long DEFAULT_TIMEOUT; static final String FS_FTP_USER_PREFIX; static final String FS_FTP_HOST; static final String FS_FTP_HOST_PORT; static final String FS_FTP_PASSWORD_PREFIX; static final String FS_FTP_DATA_CONNECTION_MODE; static final String FS_FTP_TRANSFER_MODE; static final String E_SAME_DIRECTORY_ONLY; static final String FS_FTP_TIMEOUT; }
@Test public void testFTPTransferMode() throws Exception { Configuration conf = new Configuration(); FTPFileSystem ftp = new FTPFileSystem(); assertEquals(FTP.BLOCK_TRANSFER_MODE, ftp.getTransferMode(conf)); conf.set(FTPFileSystem.FS_FTP_TRANSFER_MODE, "STREAM_TRANSFER_MODE"); assertEquals(FTP.STREAM_TRANSFER_MODE, ftp.getTransferMode(conf)); conf.set(FTPFileSystem.FS_FTP_TRANSFER_MODE, "COMPRESSED_TRANSFER_MODE"); assertEquals(FTP.COMPRESSED_TRANSFER_MODE, ftp.getTransferMode(conf)); conf.set(FTPFileSystem.FS_FTP_TRANSFER_MODE, "invalid"); assertEquals(FTPClient.BLOCK_TRANSFER_MODE, ftp.getTransferMode(conf)); }