@Override
  public void handlePurgeRequest(NetworkRequest request) throws IOException, InterruptedException {
    PurgeRequest purgeRequest = PurgeRequest.readFrom(new DataInputStream(request.getInputStream()), clusterMap);
    long requestQueueTime = SystemTime.getInstance().milliseconds() - request.getStartTimeInMs();
    long totalTimeSpent = requestQueueTime;
    long startTime = SystemTime.getInstance().milliseconds();
    PurgeResponse response = null;
    try {
      StoreKey convertedStoreKey = getConvertedStoreKeys(Collections.singletonList(purgeRequest.getBlobId())).get(0);
      ServerErrorCode error =
          validateRequest(purgeRequest.getBlobId().getPartition(), RequestOrResponseType.PurgeRequest, false);
      if (error != ServerErrorCode.NoError) {
        logger.error("Validating purge request failed with error {} for request {}", error, purgeRequest);
        response = new PurgeResponse(purgeRequest.getCorrelationId(), purgeRequest.getClientId(), error);
      } else {
        BlobId convertedBlobId = (BlobId) convertedStoreKey;
        MessageInfo info = new MessageInfo.Builder(convertedBlobId, -1, convertedBlobId.getAccountId(),
            convertedBlobId.getContainerId(), purgeRequest.getPurgeTimeInMs()).isDeleted(true)
            .lifeVersion(MessageInfo.LIFE_VERSION_FROM_FRONTEND)
            .build();
        Store storeToPurge = storeManager.getStore(purgeRequest.getBlobId().getPartition());
        storeToPurge.purge(Collections.singletonList(info));
        response =
            new PurgeResponse(purgeRequest.getCorrelationId(), purgeRequest.getClientId(), ServerErrorCode.NoError);
        if (notification != null) {
          notification.onBlobReplicaPurged(currentNode.getHostname(), currentNode.getPort(), convertedStoreKey.getID(),
              BlobReplicaSourceType.PRIMARY);
        }
      }
    } catch (StoreException e) {
      boolean logInErrorLevel = false;
      if (e.getErrorCode() == StoreErrorCodes.IDNotFound) {
        metrics.idNotFoundError.inc();
      } else if (e.getErrorCode() == StoreErrorCodes.IDPurged) {
        metrics.idPurgedError.inc();
      } else if (e.getErrorCode() == StoreErrorCodes.AuthorizationFailure) {
        metrics.purgeAuthorizationFailure.inc();
      } else {
        logInErrorLevel = true;
        metrics.unExpectedStorePurgeError.inc();
      }
      if (logInErrorLevel) {
        logger.error("Store exception on a purge message with error code {} for request {}", e.getErrorCode(),
            purgeRequest, e);
      } else {
        logger.trace("Store exception on a purge with error code {} for request {}", e.getErrorCode(), purgeRequest,
            e);
      }
      response = new PurgeResponse(purgeRequest.getCorrelationId(), purgeRequest.getClientId(),
          ErrorMapping.getStoreErrorMapping(e.getErrorCode()));
    } catch (Exception e) {
      logger.error("Unknown exception for purge request {}", purgeRequest, e);
      response = new PurgeResponse(purgeRequest.getCorrelationId(), purgeRequest.getClientId(),
          ServerErrorCode.UnknownError);
      metrics.unExpectedStorePurgeError.inc();
    } finally {
      long processingTime = SystemTime.getInstance().milliseconds() - startTime;
      totalTimeSpent += processingTime;
      publicAccessLogger.info("{} {} processingTime {}", purgeRequest, response, processingTime);
      // Update request metrics.
      RequestMetricsUpdater metricsUpdater = new RequestMetricsUpdater(requestQueueTime, processingTime, 0, 0, false);
      purgeRequest.accept(metricsUpdater);
    }
    requestResponseChannel.sendResponse(response, request,
        new ServerNetworkResponseMetrics(metrics.purgeBlobResponseQueueTimeInMs, metrics.purgeBlobSendTimeInMs,
            metrics.purgeBlobTotalTimeInMs, null, null, totalTimeSpent));
  }
Handler for all metrics to be updated for batch delete operation.
ambry-server/src/main/java/com/github/ambry/server/AmbryRequests.java