HADOOP-17511 restore xfer manager integration

Change-Id: Ifd1b217861c646a9c17b56614bd1f6da570e3d15

Author: steveloughran

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

@@ -79,8 +79,6 @@
 import com.amazonaws.services.s3.transfer.model.CopyResult;
 import com.amazonaws.services.s3.transfer.model.UploadResult;
 import com.amazonaws.event.ProgressListener;
-
-import org.apache.hadoop.fs.statistics.IOStatisticsLogging;
 import org.apache.hadoop.thirdparty.com.google.common.annotations.VisibleForTesting;
 import org.apache.hadoop.thirdparty.com.google.common.base.Preconditions;
 import org.slf4j.Logger;
@@ -133,6 +131,7 @@
 import org.apache.hadoop.fs.statistics.DurationTracker;
 import org.apache.hadoop.fs.statistics.DurationTrackerFactory;
 import org.apache.hadoop.fs.statistics.IOStatistics;
+import org.apache.hadoop.fs.statistics.IOStatisticsLogging;
 import org.apache.hadoop.fs.statistics.IOStatisticsSource;
 import org.apache.hadoop.io.IOUtils;
 import org.apache.hadoop.io.Text;
@@ -4105,7 +4104,8 @@ private CopyResult copyFile(String srcKey, String dstKey, long size,
               getRequestFactory().newCopyObjectRequest(srcKey, dstKey, srcom);
           changeTracker.maybeApplyConstraint(copyObjectRequest);
           incrementStatistic(OBJECT_COPY_REQUESTS);
-          Copy copy = transfers.copy(copyObjectRequest);
+          Copy copy = transfers.copy(copyObjectRequest,
+              getAuditManager().createStateChangeListener());
           copy.addProgressListener(progressListener);
           CopyOutcome copyOutcome = CopyOutcome.waitForCopy(copy);
           InterruptedException interruptedException =

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

@@ -21,6 +21,7 @@
 import java.util.List;
 
 import com.amazonaws.handlers.RequestHandler2;
+import com.amazonaws.services.s3.transfer.internal.TransferStateChangeListener;
 
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.service.Service;
@@ -52,4 +53,14 @@ public interface AuditManager extends Service, AuditSpanSource,
    */
   List<RequestHandler2> createRequestHandlers();
 
+  /**
+   * Return a transfer state change callback which
+   * fixes the active span context to be that in which
+   * the state change listener was created.
+   * This can be used to audit the creation of the multipart
+   * upload initiation request which the transfer manager
+   * makes when a file to be copied is split up.
+   * @return a state change listener.
+   */
+  TransferStateChangeListener createStateChangeListener();
 }

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/audit/impl/ActiveAuditManager.java

@@ -28,6 +28,8 @@
 import com.amazonaws.Response;
 import com.amazonaws.SdkBaseException;
 import com.amazonaws.handlers.RequestHandler2;
+import com.amazonaws.services.s3.transfer.Transfer;
+import com.amazonaws.services.s3.transfer.internal.TransferStateChangeListener;
 import org.apache.hadoop.thirdparty.com.google.common.base.Preconditions;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -209,6 +211,18 @@ public List<RequestHandler2> createRequestHandlers() {
     return requestHandlers;
   }
 
+  @Override
+  public TransferStateChangeListener createStateChangeListener() {
+    final AuditSpan span = getActiveThreadSpan();
+    return new TransferStateChangeListener() {
+      @Override
+      public void transferStateChanged(final Transfer transfer,
+          final Transfer.TransferState state) {
+        setActiveThreadSpan(span);
+      }
+    };
+  }
+
   /**
    * Attach a reference to the active thread span, then
    * invoke the same callback on that active thread.

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/audit/impl/NoopAuditManager.java

@@ -24,6 +24,8 @@
 import java.util.List;
 
 import com.amazonaws.handlers.RequestHandler2;
+import com.amazonaws.services.s3.transfer.Transfer;
+import com.amazonaws.services.s3.transfer.internal.TransferStateChangeListener;
 
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.fs.s3a.audit.AuditManager;
@@ -61,4 +63,12 @@ public List<RequestHandler2> createRequestHandlers() {
     return new ArrayList<>();
   }
 
+  @Override
+  public TransferStateChangeListener createStateChangeListener() {
+    return new TransferStateChangeListener() {
+      public void transferStateChanged(final Transfer transfer,
+          final Transfer.TransferState state) {
+      }
+    };
+  }
 }