@VisibleForTesting
  void setDataConnectionMode(FTPClient client, Configuration conf)
      throws IOException {
    final String mode = conf.get(FS_FTP_DATA_CONNECTION_MODE);
    if (mode == null) {
      return;
    }
    final String upper = mode.toUpperCase();
    if (upper.equals("PASSIVE_LOCAL_DATA_CONNECTION_MODE")) {
      client.enterLocalPassiveMode();
    } else if (upper.equals("PASSIVE_REMOTE_DATA_CONNECTION_MODE")) {
      client.enterRemotePassiveMode();
    } else {
      if (!upper.equals("ACTIVE_LOCAL_DATA_CONNECTION_MODE")) {
        LOG.warn("Cannot parse the value for " + FS_FTP_DATA_CONNECTION_MODE
            + ": " + mode + ". Using default.");
      }
    }
  }
@Test
  public void testFTPDataConnectionMode() throws Exception {
    Configuration conf = new Configuration();
    FTPClient client = new FTPClient();
    FTPFileSystem ftp = new FTPFileSystem();
    assertEquals(FTPClient.ACTIVE_LOCAL_DATA_CONNECTION_MODE,
        client.getDataConnectionMode());

    ftp.setDataConnectionMode(client, conf);
    assertEquals(FTPClient.ACTIVE_LOCAL_DATA_CONNECTION_MODE,
        client.getDataConnectionMode());

    conf.set(FTPFileSystem.FS_FTP_DATA_CONNECTION_MODE, "invalid");
    ftp.setDataConnectionMode(client, conf);
    assertEquals(FTPClient.ACTIVE_LOCAL_DATA_CONNECTION_MODE,
        client.getDataConnectionMode());

    conf.set(FTPFileSystem.FS_FTP_DATA_CONNECTION_MODE,
        "PASSIVE_LOCAL_DATA_CONNECTION_MODE");
    ftp.setDataConnectionMode(client, conf);
    assertEquals(FTPClient.PASSIVE_LOCAL_DATA_CONNECTION_MODE,
        client.getDataConnectionMode());

  }