Enable S3BatchDelete (#2965)

This PR implements S3BatchDelete API. Each requests takes multiple keys to delete and a returns a batch response including the status of delete - success or fail - for each key.

Co-authored-by: Allen Averbukh <aaverbukh@linkedin.com>

ambry-api/src/main/java/com/github/ambry/frontend/NamedBlobPath.java

@@ -93,6 +93,7 @@ public static NamedBlobPath parseS3(String path, Map<String, Object> args) throw
     path = path.startsWith("/") ? path.substring(1) : path;
     String[] splitPath = path.split("/", 4);
     String blobNamePrefix = RestUtils.getHeader(args, PREFIX_PARAM, false);
+    boolean isBatchDelete = args.containsKey(BATCH_DELETE_QUERY_PARAM);
     boolean isGetObjectLockRequest = args.containsKey(OBJECT_LOCK_PARAM);
     //There are two cases for S3 listing
     //1.has prefix (Ex:GET /?prefix=prefixName&delimiter=&encoding-type=url)
@@ -118,7 +119,7 @@ public static NamedBlobPath parseS3(String path, Map<String, Object> args) throw
       }
       return new NamedBlobPath(accountName, containerName, null, blobNamePrefix, pageToken);
     }
-    if (isGetObjectLockRequest) {
+    if (isGetObjectLockRequest || isBatchDelete) {
       return new NamedBlobPath(accountName, containerName, null, null, null);
     } else {
       String blobName = splitPath[3];

ambry-api/src/main/java/com/github/ambry/frontend/s3/S3MessagePayload.java

@@ -14,9 +14,12 @@
  */
 package com.github.ambry.frontend.s3;
 
+import java.util.ArrayList;
 import java.util.List;
 import com.fasterxml.jackson.dataformat.xml.annotation.JacksonXmlElementWrapper;
+import com.fasterxml.jackson.dataformat.xml.annotation.JacksonXmlRootElement;
 import com.fasterxml.jackson.dataformat.xml.annotation.JacksonXmlProperty;
+import java.util.concurrent.ConcurrentLinkedQueue;
 
 
 /**
@@ -351,4 +354,143 @@ public String toString() {
       return "Bucket=" + bucket + ", Key=" + key + ", UploadId=" + uploadId;
     }
   }
+
+  public static class S3BatchDeleteObjects {
+
+    // Ensure that the "Delete" wrapper element is mapped correctly to the list of "Object" elements
+    @JacksonXmlElementWrapper(useWrapping = false)  // Avoids wrapping the <Delete> element itself
+    @JacksonXmlProperty(localName = "Object")      // Specifies that each <Object> element maps to an instance of S3BatchDeleteKeys
+    private List<S3BatchDeleteKey> objects;
+
+    public List<S3BatchDeleteKey> getObjects() {
+      return objects;
+    }
+
+    public void setObjects(List<S3BatchDeleteKey> objects) {
+      this.objects = objects;
+    }
+
+    @Override
+    public String toString() {
+      return "S3BatchDeleteObjects{" +
+          "objects=" + objects +
+          '}';
+    }
+  }
+
+  public static class S3BatchDeleteKey {
+
+    // Maps the <Key> element inside each <Object> to the 'key' property in S3BatchDeleteKeys
+    @JacksonXmlProperty(localName = "Key")
+    private String key;
+
+    public String getKey() {
+      return key;
+    }
+
+    public void setKey(String key) {
+      this.key = key;
+    }
+
+    @Override
+    public String toString() {
+      return "S3BatchDeleteKey{" +
+          "key='" + key + '\'' +
+          '}';
+    }
+  }
+
+  // exact naming from S3BatchDelete response
+  public static class DeleteResult {
+
+    @JacksonXmlElementWrapper(useWrapping = false)
+    @JacksonXmlProperty(localName = "Error") // Maps to <Error> in XML
+    private List<S3MessagePayload.S3ErrorObject> errors;
+
+    @JacksonXmlElementWrapper(useWrapping = false)
+    @JacksonXmlProperty(localName = "Deleted") // Maps to <Deleted> in XML
+    private List<S3MessagePayload.S3DeletedObject> deleted;
+
+    // Getters and setters
+    public List<S3MessagePayload.S3ErrorObject> getErrors() {
+      return errors;
+    }
+
+    public void setErrors(List<S3MessagePayload.S3ErrorObject> errors) {
+      this.errors = errors;
+    }
+
+    public List<S3MessagePayload.S3DeletedObject> getDeleted() {
+      return deleted;
+    }
+
+    public void setDeleted(List<S3MessagePayload.S3DeletedObject> deleted) {
+      this.deleted = deleted;
+    }
+  }
+
+  public static class S3ErrorObject {
+
+    @JacksonXmlProperty(localName = "Key")
+    private String key;
+
+    @JacksonXmlProperty(localName = "Code")
+    private String code;
+
+    public S3ErrorObject(){}
+
+    public S3ErrorObject(String key, String code) {
+      this.key = key;
+      this.code = code;
+    }
+
+    // Getters and setters
+    public String getKey() {
+      return key;
+    }
+
+    public void setKey(String key) {
+      this.key = key;
+    }
+
+    public String getCode() {
+      return code;
+    }
+
+    public void setCode(String code) {
+      this.code = code;
+    }
+
+    @Override
+    public String toString() {
+      return "S3ErrorObject{key='" + key + "', code='" + code + "'}";
+    }
+  }
+
+  public static class S3DeletedObject {
+
+    @JacksonXmlProperty(localName = "Key")
+    private String key;
+
+    public S3DeletedObject() {}
+
+    public S3DeletedObject(String key) {
+      this.key = key;
+    }
+
+    // Getters and setters
+    public String getKey() {
+      return key;
+    }
+
+    public void setKey(String key) {
+      this.key = key;
+    }
+
+    @Override
+    public String toString() {
+      return "S3DeletedObject{key='" + key + "'}";
+    }
+  }
 }
+

ambry-api/src/main/java/com/github/ambry/rest/RestUtils.java

@@ -66,6 +66,7 @@ public class RestUtils {
   public static final String PATH_SEPARATOR_STRING = "/";
   public static final String STITCH = "STITCH";
   public static final String UPLOADS_QUERY_PARAM = "uploads";
+  public static final String BATCH_DELETE_QUERY_PARAM = "delete";
   public static final String UPLOAD_ID_QUERY_PARAM = "uploadId";
   public static final String CONTINUE = "100-continue";
   public static final String OBJECT_LOCK_PARAM = "object-lock";

ambry-frontend/src/main/java/com/github/ambry/frontend/FrontendMetrics.java

@@ -18,6 +18,7 @@
 import com.codahale.metrics.Meter;
 import com.codahale.metrics.MetricRegistry;
 import com.github.ambry.config.FrontendConfig;
+import com.github.ambry.frontend.s3.S3BatchDeleteHandler;
 import com.github.ambry.frontend.s3.S3DeleteHandler;
 import com.github.ambry.frontend.s3.S3GetHandler;
 import com.github.ambry.frontend.s3.S3ListHandler;
@@ -35,6 +36,7 @@ public class FrontendMetrics {
   // RestRequestMetricsGroup
   // DELETE
   public final RestRequestMetricsGroup deleteBlobMetricsGroup;
+  public final RestRequestMetricsGroup batchDeleteMetricsGroup;
   public final RestRequestMetricsGroup deleteDatasetsMetricsGroup;
   //COPY
   public final RestRequestMetricsGroup copyBlobMetricsGroup;
@@ -165,6 +167,7 @@ public class FrontendMetrics {
   public final AsyncOperationTracker.Metrics deleteDatasetOutOfRetentionRequestMetrics;
 
   public final AsyncOperationTracker.Metrics s3DeleteHandleMetrics;
+  public final AsyncOperationTracker.Metrics s3BatchDeleteHandleMetrics;
   public final AsyncOperationTracker.Metrics s3ListHandleMetrics;
   public final AsyncOperationTracker.Metrics s3PutHandleMetrics;
   public final AsyncOperationTracker.Metrics s3GetHandleMetrics;
@@ -315,6 +318,9 @@ public FrontendMetrics(MetricRegistry metricRegistry, FrontendConfig frontendCon
     deleteBlobMetricsGroup =
         new RestRequestMetricsGroup(FrontendRestRequestService.class, "DeleteBlob", false, metricRegistry,
             frontendConfig);
+    batchDeleteMetricsGroup =
+        new RestRequestMetricsGroup(FrontendRestRequestService.class, "BatchDeleteBlob", false, metricRegistry,
+            frontendConfig);
     deleteDatasetsMetricsGroup =
         new RestRequestMetricsGroup(FrontendRestRequestService.class, "DeleteDataset", false, metricRegistry,
             frontendConfig);
@@ -505,6 +511,7 @@ public FrontendMetrics(MetricRegistry metricRegistry, FrontendConfig frontendCon
         new AsyncOperationTracker.Metrics(NamedBlobPutHandler.class, "RetentionRequest", metricRegistry);
 
     s3DeleteHandleMetrics = new AsyncOperationTracker.Metrics(S3DeleteHandler.class, "S3Handle", metricRegistry);
+    s3BatchDeleteHandleMetrics = new AsyncOperationTracker.Metrics(S3BatchDeleteHandler.class, "S3Handle", metricRegistry);
     s3ListHandleMetrics = new AsyncOperationTracker.Metrics(S3ListHandler.class, "S3Handle", metricRegistry);
     s3PutHandleMetrics = new AsyncOperationTracker.Metrics(S3PutHandler.class, "S3Handle", metricRegistry);
     s3GetHandleMetrics = new AsyncOperationTracker.Metrics(S3GetHandler.class, "S3Handle", metricRegistry);

ambry-frontend/src/main/java/com/github/ambry/frontend/FrontendRestRequestService.java

@@ -18,6 +18,7 @@
 import com.github.ambry.clustermap.ClusterMap;
 import com.github.ambry.commons.Callback;
 import com.github.ambry.config.FrontendConfig;
+import com.github.ambry.frontend.s3.S3BatchDeleteHandler;
 import com.github.ambry.frontend.s3.S3DeleteHandler;
 import com.github.ambry.frontend.s3.S3GetHandler;
 import com.github.ambry.frontend.s3.S3HeadHandler;
@@ -113,6 +114,7 @@ class FrontendRestRequestService implements RestRequestService {
   private PostDatasetsHandler postDatasetsHandler;
   private GetStatsReportHandler getStatsReportHandler;
   private S3DeleteHandler s3DeleteHandler;
+  private S3BatchDeleteHandler s3BatchDeleteHandler;
   private S3ListHandler s3ListHandler;
   private S3PutHandler s3PutHandler;
   private S3HeadHandler s3HeadHandler;
@@ -240,7 +242,8 @@ public void start() throws InstantiationException {
         new S3MultipartUploadHandler(securityService, frontendMetrics, accountAndContainerInjector, frontendConfig,
             namedBlobDb, idConverter, router, quotaManager);
     s3DeleteHandler = new S3DeleteHandler(deleteBlobHandler, s3MultipartUploadHandler, frontendMetrics);
-    s3PostHandler = new S3PostHandler(s3MultipartUploadHandler);
+    s3BatchDeleteHandler = new S3BatchDeleteHandler(deleteBlobHandler, frontendMetrics);
+    s3PostHandler = new S3PostHandler(s3MultipartUploadHandler, s3BatchDeleteHandler);
     s3PutHandler = new S3PutHandler(namedBlobPutHandler, s3MultipartUploadHandler, frontendMetrics);
     s3ListHandler = new S3ListHandler(namedBlobListHandler, frontendMetrics);
     s3GetHandler =

ambry-frontend/src/main/java/com/github/ambry/frontend/s3/S3BaseHandler.java

@@ -44,7 +44,7 @@
  *            require a response body, otherwise it is {@link ReadableStreamChannel}.
  */
 abstract public class S3BaseHandler<R> {
-  private static final Logger LOGGER = LoggerFactory.getLogger(S3BaseHandler.class);
+  protected static final Logger LOGGER = LoggerFactory.getLogger(S3BaseHandler.class);
 
   /**
    * Handles the S3 request and construct the response.
@@ -122,6 +122,15 @@ public static boolean isMultipartCreateUploadRequest(RestRequest restRequest) {
         && restRequest.getArgs().containsKey(UPLOADS_QUERY_PARAM);
   }
 
+  /**
+   * @param restRequest the {@link RestRequest} that contains the request parameters.
+   * @return {@code True} if it is a request for batch delete.
+   */
+  public static boolean isBatchDelete(RestRequest restRequest) {
+    return restRequest.getRestMethod() == RestMethod.POST && restRequest.getArgs().containsKey(S3_REQUEST)
+        && restRequest.getArgs().containsKey(BATCH_DELETE_QUERY_PARAM);
+  }
+
   /**
    * @param restRequest the {@link RestRequest} that contains the request parameters.
    * @return {@code True} if it is a completion/abortion of multipart uploads.