private Callback<Long> parseRequestBodyAndDeleteCallback(RetainingAsyncWritableChannel channel,
      RestRequest restRequest, DeleteBlobHandler deleteBlobHandler, RestResponseChannel restResponseChannel,
      Callback<ReadableStreamChannel> finalCallback) {

    return buildCallback(metrics.s3BatchDeleteHandleMetrics, bytesRead -> {
      S3MessagePayload.S3BatchDeleteObjects deleteRequest = null;
      try {
        // check for empty message
        if (bytesRead == 0) {
          logger.trace("bytesRead is empty");
          throw new RestServiceException("bytesRead is empty", RestServiceErrorCode.BadRequest);
        }

        // ensure request body format is correct
        deleteRequest = deserializeRequest(channel);
        if (deleteRequest.getObjects() == null || deleteRequest.getObjects().isEmpty()) {
          logger.trace("s3batchdelete request size needs to be at least 1");
          String batchSizeEqualsZero = "Request size needs to be at least 1";
          throw new RestServiceException(batchSizeEqualsZero, RestServiceErrorCode.BadRequest);
        }

        // validate the request for size
        if (deleteRequest.getObjects().size() > MAX_BATCH_DELETE_SIZE) {
          logger.trace("exceeded max batch delete size " + MAX_BATCH_DELETE_SIZE);
          String batchSizeErrorMessage = "Exceeded Maximum Batch Size of " + MAX_BATCH_DELETE_SIZE;
          throw new RestServiceException(batchSizeErrorMessage, RestServiceErrorCode.BadRequest);
        }
      } catch (Exception e) {
        S3MessagePayload.Error response = new S3MessagePayload.Error();
        response.setCode(ERR_MALFORMED_REQUEST_BODY_CODE);
        response.setMessage(ERR_MALFORMED_REQUEST_BODY_MESSAGE);
        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
        xmlMapper.writeValue(outputStream, response);
        ReadableStreamChannel readableStreamChannel = new ByteBufferReadableStreamChannel(ByteBuffer.wrap(outputStream.toByteArray()));
        restResponseChannel.setHeader(Headers.CONTENT_LENGTH, readableStreamChannel.getSize());
        restResponseChannel.setHeader(Headers.CONTENT_TYPE, XML_CONTENT_TYPE);
        restResponseChannel.setStatus(ResponseStatus.Ok);
        finalCallback.onCompletion(readableStreamChannel, null);
        return;
      }

      RequestPath requestPath = (RequestPath) restRequest.getArgs().get(InternalKeys.REQUEST_PATH);

      // create objects for processing the request
      ConcurrentLinkedQueue<S3MessagePayload.S3ErrorObject> errors = new ConcurrentLinkedQueue<>();
      ConcurrentLinkedQueue<S3MessagePayload.S3DeletedObject> deleted = new ConcurrentLinkedQueue<>();
      List<CompletableFuture<Void>> deleteFutures = new ArrayList<>();

      for (S3MessagePayload.S3BatchDeleteKey object : deleteRequest.getObjects()) {
        String singleDeletePath = requestPath.getPathAfterPrefixes() + "/" + object.getKey();
        RequestPath newRequestPath =
            RequestPath.parse(singleDeletePath, restRequest.getArgs(), new ArrayList<>(), requestPath.getClusterName());
        WrappedRestRequest singleDeleteRequest = new WrappedRestRequest(restRequest);
        singleDeleteRequest.setArg(InternalKeys.REQUEST_PATH, newRequestPath);
        NoOpResponseChannel noOpResponseChannel = new NoOpResponseChannel();
        CompletableFuture<Void> future = new CompletableFuture<>();

        // Handle the delete operation using the deleteBlobHandler
        deleteBlobHandler.handle(singleDeleteRequest, noOpResponseChannel, (result, exception) -> {
          // Call our custom onDeleteCompletion to track success/failure
          if (exception == null) {
            deleted.add(new S3MessagePayload.S3DeletedObject(object.getKey()));
          } else if (exception instanceof RestServiceException) {
            RestServiceException restServiceException = (RestServiceException) exception;
            errors.add((new S3MessagePayload.S3ErrorObject(object.getKey(), restServiceException.getErrorCode().toString())));
          } else {
            errors.add((new S3MessagePayload.S3ErrorObject(object.getKey(), RestServiceErrorCode.InternalServerError.toString())));
          }
          future.complete(null);
        });
        deleteFutures.add(future);
      }

      CompletableFuture.allOf(deleteFutures.toArray(new CompletableFuture[0]))
          .whenComplete((result, exception) -> {
            try {
              // construct and serialize response
              S3MessagePayload.DeleteResult response = new S3MessagePayload.DeleteResult();
              ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
              response.setDeleted(new ArrayList<>(deleted));
              response.setErrors(new ArrayList<>(errors));
              xmlMapper.writeValue(outputStream, response);
              ReadableStreamChannel readableStreamChannel = new ByteBufferReadableStreamChannel(ByteBuffer.wrap(outputStream.toByteArray()));
              restResponseChannel.setHeader(Headers.CONTENT_LENGTH, readableStreamChannel.getSize());
              restResponseChannel.setHeader(Headers.CONTENT_TYPE, XML_CONTENT_TYPE);
              restResponseChannel.setStatus(ResponseStatus.Ok);
              finalCallback.onCompletion(readableStreamChannel, null);
            } catch (IOException | RestServiceException e) {
              finalCallback.onCompletion(null, e);
            }
      });
    }, restRequest.getUri(), LOGGER, finalCallback);
  }
This method will read the request body, deserialize it, and trigger the batch delete process.
ambry-frontend/src/main/java/com/github/ambry/frontend/s3/S3BatchDeleteHandler.java