From 24dc068a361648b4e59e1807b07ff2239f41c740 Mon Sep 17 00:00:00 2001
From: Adam Antal <adam.antal@cloudera.com>
Date: Wed, 17 Oct 2018 11:32:17 -0700
Subject: [PATCH] HADOOP-11100. Support to configure
 ftpClient.setControlKeepAliveTimeout. Contributed by Adam Antal.

Signed-off-by: Xiao Chen <xiao@apache.org>
---
 .../org/apache/hadoop/fs/ftp/FTPFileSystem.java   | 13 +++++++++++++
 .../src/main/resources/core-default.xml           |  8 ++++++++
 .../conf/TestCommonConfigurationFields.java       |  1 +
 .../apache/hadoop/fs/ftp/TestFTPFileSystem.java   | 15 +++++++++++++++
 4 files changed, 37 insertions(+)

diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/ftp/FTPFileSystem.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/ftp/FTPFileSystem.java
index 676c207e00..4b144bfddf 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/ftp/FTPFileSystem.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/ftp/FTPFileSystem.java
@@ -62,6 +62,7 @@ public class FTPFileSystem extends FileSystem {
   public static final int DEFAULT_BUFFER_SIZE = 1024 * 1024;
 
   public static final int DEFAULT_BLOCK_SIZE = 4 * 1024;
+  public static final long DEFAULT_TIMEOUT = 0;
   public static final String FS_FTP_USER_PREFIX = "fs.ftp.user.";
   public static final String FS_FTP_HOST = "fs.ftp.host";
   public static final String FS_FTP_HOST_PORT = "fs.ftp.host.port";
@@ -71,6 +72,7 @@ public class FTPFileSystem extends FileSystem {
   public static final String FS_FTP_TRANSFER_MODE = "fs.ftp.transfer.mode";
   public static final String E_SAME_DIRECTORY_ONLY =
       "only same directory renames are supported";
+  public static final String FS_FTP_TIMEOUT = "fs.ftp.timeout";
 
   private URI uri;
 
@@ -150,6 +152,7 @@ private FTPClient connect() throws IOException {
       client.setFileTransferMode(getTransferMode(conf));
       client.setFileType(FTP.BINARY_FILE_TYPE);
       client.setBufferSize(DEFAULT_BUFFER_SIZE);
+      setTimeout(client, conf);
       setDataConnectionMode(client, conf);
     } else {
       throw new IOException("Login failed on server - " + host + ", port - "
@@ -159,6 +162,16 @@ private FTPClient connect() throws IOException {
     return client;
   }
 
+  /**
+   * Set the FTPClient's timeout based on configuration.
+   * FS_FTP_TIMEOUT is set as timeout (defaults to DEFAULT_TIMEOUT).
+   */
+  @VisibleForTesting
+  void setTimeout(FTPClient client, Configuration conf) {
+    long timeout = conf.getLong(FS_FTP_TIMEOUT, DEFAULT_TIMEOUT);
+    client.setControlKeepAliveTimeout(timeout);
+  }
+
   /**
    * Set FTP's transfer mode based on configuration. Valid values are
    * STREAM_TRANSFER_MODE, BLOCK_TRANSFER_MODE and COMPRESSED_TRANSFER_MODE.
diff --git a/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml b/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml
index 32dd622894..599396f9da 100644
--- a/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml
+++ b/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml
@@ -925,6 +925,14 @@
   </description>
 </property>
 
+<property>
+  <name>fs.ftp.timeout</name>
+  <value>0</value>
+  <description>
+    FTP filesystem's timeout in seconds.
+  </description>
+</property>
+
 <property>
   <name>fs.df.interval</name>
   <value>60000</value>
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/conf/TestCommonConfigurationFields.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/conf/TestCommonConfigurationFields.java
index e10617daab..2766b5605b 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/conf/TestCommonConfigurationFields.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/conf/TestCommonConfigurationFields.java
@@ -93,6 +93,7 @@ public void initializeMemberVariables() {
     xmlPropsToSkipCompare.add("fs.ftp.user.localhost");
     xmlPropsToSkipCompare.add("fs.ftp.data.connection.mode");
     xmlPropsToSkipCompare.add("fs.ftp.transfer.mode");
+    xmlPropsToSkipCompare.add("fs.ftp.timeout");
     xmlPropsToSkipCompare.add("hadoop.tmp.dir");
     xmlPropsToSkipCompare.add("nfs3.mountd.port");
     xmlPropsToSkipCompare.add("nfs3.server.port");
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/ftp/TestFTPFileSystem.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/ftp/TestFTPFileSystem.java
index aab52ae28e..3d41ccb91d 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/ftp/TestFTPFileSystem.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/ftp/TestFTPFileSystem.java
@@ -137,4 +137,19 @@ private FTPFile getFTPFileOf(int access, FsAction action) {
     return ftpFile;
   }
 
+  @Test
+  public void testFTPSetTimeout() {
+    Configuration conf = new Configuration();
+    FTPClient client = new FTPClient();
+    FTPFileSystem ftp = new FTPFileSystem();
+
+    ftp.setTimeout(client, conf);
+    assertEquals(client.getControlKeepAliveTimeout(),
+        FTPFileSystem.DEFAULT_TIMEOUT);
+
+    long timeout = 600;
+    conf.setLong(FTPFileSystem.FS_FTP_TIMEOUT, timeout);
+    ftp.setTimeout(client, conf);
+    assertEquals(client.getControlKeepAliveTimeout(), timeout);
+  }
 }
\ No newline at end of file