@Override
  public void handleBatchDeleteRequest(NetworkRequest request) throws IOException, InterruptedException {
    BatchDeleteRequest batchDeleteRequest = BatchDeleteRequest.readFrom(new DataInputStream(request.getInputStream()), clusterMap);
    logger.trace("Batch delete request {} received.", batchDeleteRequest);
    long requestQueueTime = SystemTime.getInstance().milliseconds() - request.getStartTimeInMs();
    long totalTimeSpent = requestQueueTime;
    long startTime = SystemTime.getInstance().milliseconds();
    ServerErrorCode operationServerErrorCode = ServerErrorCode.NoError;
    List<BatchDeletePartitionResponseInfo> partitionResponseInfoList = new ArrayList<>();
    BatchDeleteResponse response = null;
    for (BatchDeletePartitionRequestInfo batchDeletePartitionRequestInfo : batchDeleteRequest.getPartitionRequestInfoList()) {
      BatchDeletePartitionResponseInfo partitionResponseInfo = null;
      try {
        ServerErrorCode error =
            validateRequest(batchDeletePartitionRequestInfo.getPartition(), RequestOrResponseType.BatchDeleteRequest,
                false);
        if (error != ServerErrorCode.NoError) {
          logger.error("Validating delete request failed with error {} for request {}", error, batchDeleteRequest);
          List<BlobDeleteStatus> blobsDeleteStatuses = new ArrayList<>();
          for (StoreKey storeKey: batchDeletePartitionRequestInfo.getBlobIds()){
            blobsDeleteStatuses.add(new BlobDeleteStatus((BlobId) storeKey, error));
          }
          partitionResponseInfo = new BatchDeletePartitionResponseInfo(batchDeletePartitionRequestInfo.getPartition(), blobsDeleteStatuses);
        } else {
          partitionResponseInfo = batchDeleteForPartition(batchDeleteRequest, batchDeletePartitionRequestInfo);
        }
      } catch (Exception e) {
        // All blobs failed to delete in a partition
        // This is when some unknown exception gets thrown while deleting a batch of blobs in a partition,
        // in that case all blobDeleteStatus will be marked Bad_Request
        logger.trace("Exception when processing Batch delete request {}", e.toString());
        List<BlobDeleteStatus> blobDeleteStatuses = new ArrayList<>();
        for (StoreKey storeKey : batchDeletePartitionRequestInfo.getBlobIds()){
          blobDeleteStatuses.add(new BlobDeleteStatus((BlobId) storeKey, ServerErrorCode.UnknownError));
        }
        partitionResponseInfo = new BatchDeletePartitionResponseInfo(batchDeletePartitionRequestInfo.getPartition(), blobDeleteStatuses);
      } finally {
        long processingTime = SystemTime.getInstance().milliseconds() - startTime;
        totalTimeSpent += processingTime;
        publicAccessLogger.info("{} {} processingTime {}", batchDeleteRequest, response, processingTime);
        // Update request metrics.
        RequestMetricsUpdater metricsUpdater = new RequestMetricsUpdater(requestQueueTime, processingTime, 0, 0, false);
        batchDeleteRequest.accept(metricsUpdater);
        operationServerErrorCode = getBatchDeleteOperationServerErrorCode(partitionResponseInfo);
        handleBatchDeleteMetrics(partitionResponseInfo, batchDeleteRequest, operationServerErrorCode);
        partitionResponseInfoList.add(partitionResponseInfo);
      }
    }
    // partitionResponseInfoList will have status for each blob in each partition.
    // There is one global operation level status which will be set to UNKNOWN_ERROR if any of the blobs fail.
    // There will also be individual blob level status for each blob in each partition in this list for blob-level handling on FE.
    // Frontend will rely on this for full success, partial success and full failure scenarios in E2E operation handling.
    response = new BatchDeleteResponse(batchDeleteRequest.getCorrelationId(), batchDeleteRequest.getClientId(), partitionResponseInfoList, operationServerErrorCode);
    logger.trace("Batch delete response {}.", response);
    requestResponseChannel.sendResponse(response, request,
    new ServerNetworkResponseMetrics(metrics.batchDeleteBlobResponseQueueTimeInMs, metrics.batchDeleteBlobSendTimeInMs,
        metrics.batchDeleteBlobTotalTimeInMs, null, null, totalTimeSpent));
  }
Handles a {@link BatchDeleteRequest} and sends out the response via the {@link RequestResponseChannel}. The input
request contains a list of {@link BatchDeletePartitionRequestInfo} each containing a list of {@link StoreKey} to be
deleted in batch. This method handles full success, partial success and full failure scenarios by forming a batch
response {@link BatchDeleteResponse} and sending it out via the {@link RequestResponseChannel}.
@param request the {@link NetworkRequest} that contains the {@link BatchDeleteRequest}.
@throws IOException if there is an I/O error reading from the {@link NetworkRequest}.
@throws InterruptedException if the operation is interrupted.
ambry-server/src/main/java/com/github/ambry/server/AmbryRequests.java