From 707773b6e14b61b31ecd5473eaafa75dd5217707 Mon Sep 17 00:00:00 2001
From: kazu <kazu@bloomreach.com>
Date: Fri, 24 Mar 2017 18:29:19 -0700
Subject: [PATCH] HADOOP-14239. S3A Retry Multiple S3 Key Deletion

---
 .../apache/hadoop/fs/s3a/S3AFileSystem.java   | 21 +++++++++++++++++--
 1 file changed, 19 insertions(+), 2 deletions(-)

diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java
index 1786e68a53a96..0c7001cc69dff 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java
@@ -59,6 +59,8 @@
 import com.amazonaws.services.s3.model.SSECustomerKey;
 import com.amazonaws.services.s3.model.UploadPartRequest;
 import com.amazonaws.services.s3.model.UploadPartResult;
+import com.amazonaws.services.s3.model.MultiObjectDeleteException;
+import com.amazonaws.services.s3.model.MultiObjectDeleteException.DeleteError;
 import com.amazonaws.services.s3.transfer.Copy;
 import com.amazonaws.services.s3.transfer.TransferManager;
 import com.amazonaws.services.s3.transfer.TransferManagerConfiguration;
@@ -1006,10 +1008,25 @@ private void blockRootDelete(String key) throws InvalidRequestException {
    * operation statistics.
    * @param deleteRequest keys to delete on the s3-backend
    */
-  private void deleteObjects(DeleteObjectsRequest deleteRequest) {
+  private void deleteObjects(DeleteObjectsRequest deleteRequest) throws InvalidRequestException {
     incrementWriteOperations();
     incrementStatistic(OBJECT_DELETE_REQUESTS, 1);
-    s3.deleteObjects(deleteRequest);
+    try {
+      s3.deleteObjects(deleteRequest);
+    } catch (MultiObjectDeleteException ex) {
+      List<DeleteObjectsRequest.KeyVersion> errorKeys = new ArrayList<>();
+      for (DeleteError deleteError : ex.getErrors()) {
+        errorKeys.add(new DeleteObjectsRequest.KeyVersion(deleteError.getKey()));
+      }
+      if (errorKeys.size() == deleteRequest.getKeys().size()) {
+        // fallback to single key deletion if all of them failed
+        for (DeleteObjectsRequest.KeyVersion keyVersion : errorKeys) {
+          deleteObject(keyVersion.getKey());
+        }
+      } else {
+        deleteObjects(deleteRequest.withKeys(errorKeys));
+      }
+    }
   }
 
   /**