HADOOP-11463 Replace method-local TransferManager object with S3AFileSystem#transfers. (Ted Yu via stevel)

hadoop-common-project/hadoop-common/CHANGES.txt

@@ -175,6 +175,9 @@ Release 2.7.0 - UNRELEASED
     HADOOP-11492. Bump up curator version to 2.7.1. (Arun Suresh and
     Karthik Kambatla via kasha)
 
+    HADOOP-11463 Replace method-local TransferManager object with
+    S3AFileSystem#transfers. (Ted Yu via stevel)
+
   OPTIMIZATIONS
 
     HADOOP-11323. WritableComparator#compare keeps reference to byte array.

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java

@@ -292,7 +292,6 @@ public class S3AFileSystem extends FileSystem {
       Date purgeBefore = new Date(new Date().getTime() - purgeExistingMultipartAge*1000);
 
       transfers.abortMultipartUploads(bucket, purgeBefore);
-      transfers.shutdownNow(false);
     }
 
     serverSideEncryptionAlgorithm = conf.get(SERVER_SIDE_ENCRYPTION_ALGORITHM);
@@ -995,13 +994,6 @@ public class S3AFileSystem extends FileSystem {
     LocalFileSystem local = getLocal(getConf());
     File srcfile = local.pathToFile(src);
 
-    TransferManagerConfiguration transferConfiguration = new TransferManagerConfiguration();
-    transferConfiguration.setMinimumUploadPartSize(partSize);
-    transferConfiguration.setMultipartUploadThreshold(partSizeThreshold);
-
-    TransferManager transfers = new TransferManager(s3);
-    transfers.setConfiguration(transferConfiguration);
-
     final ObjectMetadata om = new ObjectMetadata();
     if (StringUtils.isNotBlank(serverSideEncryptionAlgorithm)) {
       om.setServerSideEncryption(serverSideEncryptionAlgorithm);
@@ -1029,8 +1021,6 @@ public class S3AFileSystem extends FileSystem {
       statistics.incrementWriteOps(1);
     } catch (InterruptedException e) {
       throw new IOException("Got interrupted, cancelling");
-    } finally {
-      transfers.shutdownNow(false);
     }
 
     // This will delete unnecessary fake parent directories
@@ -1041,6 +1031,18 @@ public class S3AFileSystem extends FileSystem {
     }
   }
 
+  @Override
+  public void close() throws IOException {
+    try {
+      super.close();
+    } finally {
+      if (transfers != null) {
+        transfers.shutdownNow(true);
+        transfers = null;
+      }
+    }
+  }
+
   /**
   * Override getCononicalServiceName because we don't support token in S3A
   */
@@ -1055,12 +1057,6 @@ public class S3AFileSystem extends FileSystem {
       LOG.debug("copyFile " + srcKey + " -> " + dstKey);
     }
 
-    TransferManagerConfiguration transferConfiguration = new TransferManagerConfiguration();
-    transferConfiguration.setMultipartCopyPartSize(partSize);
-
-    TransferManager transfers = new TransferManager(s3);
-    transfers.setConfiguration(transferConfiguration);
-
     ObjectMetadata srcom = s3.getObjectMetadata(bucket, srcKey);
     final ObjectMetadata dstom = srcom.clone();
     if (StringUtils.isNotBlank(serverSideEncryptionAlgorithm)) {
@@ -1089,8 +1085,6 @@ public class S3AFileSystem extends FileSystem {
       statistics.incrementWriteOps(1);
     } catch (InterruptedException e) {
       throw new IOException("Got interrupted, cancelling");
-    } finally {
-      transfers.shutdownNow(false);
     }
   }