@VisibleForTesting
  int getTransferMode(Configuration conf) {
    final String mode = conf.get(FS_FTP_TRANSFER_MODE);
    // FTP default is STREAM_TRANSFER_MODE, but Hadoop FTPFS's default is
    // FTP.BLOCK_TRANSFER_MODE historically.
    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;
  }
@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));
  }