@Override
  public void handleUndeleteRequest(NetworkRequest request) {

    if (!(request instanceof LocalRequestResponseChannel.LocalChannelRequest)) {
      throw new IllegalArgumentException("The request must be of LocalChannelRequest type");
    }

    UndeleteRequest undeleteRequest;

    // This is a case where handleUndeleteRequest is called when frontends are talking to Azure. In this case, this method
    // is called by request handler threads running within the frontend router itself. So, the request can be directly
    // referenced as java objects without any need for deserialization.
    undeleteRequest =
        (UndeleteRequest) ((LocalRequestResponseChannel.LocalChannelRequest) request).getRequestInfo().getRequest();

    AtomicReference<UndeleteResponse> response = new AtomicReference<>();

    ServerErrorCode error =
        validateRequest(undeleteRequest.getBlobId().getPartition(), RequestOrResponseType.UndeleteRequest, false);
    if (error != ServerErrorCode.NoError) {
      logger.error("Validating undelete request failed with error {} for request {}", error, undeleteRequest);
      response.set(new UndeleteResponse(undeleteRequest.getCorrelationId(), undeleteRequest.getClientId(), error));
    } else {
      BlobId blobId = undeleteRequest.getBlobId();
      MessageInfo info = new MessageInfo.Builder(blobId, -1, blobId.getAccountId(), blobId.getContainerId(),
          undeleteRequest.getOperationTimeMs()).isUndeleted(true)
          .lifeVersion(MessageInfo.LIFE_VERSION_FROM_FRONTEND)
          .build();
      CloudBlobStore cloudBlobStore =
          (CloudBlobStore) storeManager.getStore(undeleteRequest.getBlobId().getPartition());
      cloudBlobStore.undeleteAsync(info).whenCompleteAsync((lifeVersion, throwable) -> {
        if (throwable != null) {
          Exception ex = Utils.extractFutureExceptionCause(throwable);
          if (ex instanceof StoreException) {
            logger.error("Store exception on a undelete with error code {} for request {}",
                ((StoreException) ex).getErrorCode(), undeleteRequest, ex);
            StoreException storeException = (StoreException) ex;
            if (storeException.getErrorCode() == StoreErrorCodes.IDUndeleted) {
              if (ex instanceof IdUndeletedStoreException) {
                response.set(new UndeleteResponse(undeleteRequest.getCorrelationId(), undeleteRequest.getClientId(),
                    ((IdUndeletedStoreException) ex).getLifeVersion(), ServerErrorCode.BlobAlreadyUndeleted));
              } else {
                response.set(new UndeleteResponse(undeleteRequest.getCorrelationId(), undeleteRequest.getClientId(),
                    MessageInfo.LIFE_VERSION_FROM_FRONTEND, ServerErrorCode.BlobAlreadyUndeleted));
              }
            } else {
              logger.error("Unknown exception for undelete request{}", undeleteRequest, ex);
              response.set(new UndeleteResponse(undeleteRequest.getCorrelationId(), undeleteRequest.getClientId(),
                  ErrorMapping.getStoreErrorMapping(storeException.getErrorCode())));
            }
          } else {
            logger.error("Unknown exception for undelete request{}", undeleteRequest, ex);
            response.set(new UndeleteResponse(undeleteRequest.getCorrelationId(), undeleteRequest.getClientId(),
                ServerErrorCode.UnknownError));
          }
        } else {
          response.set(
              new UndeleteResponse(undeleteRequest.getCorrelationId(), undeleteRequest.getClientId(), lifeVersion));
        }
        try {
          requestResponseChannel.sendResponse(response.get(), request, null);
        } catch (InterruptedException ie) {
          logger.warn("Interrupted while enqueuing the response", ie);
        }
      });
    }
  }
@param storeManager {@link StoreManager} object to get stores for replicas
@param requestResponseChannel the {@link RequestResponseChannel} to receive requests and send responses.
@param clusterMap the {@link ClusterMap} of the cluster
@param nodeId the data node Id.
@param registry the {@link MetricRegistry}
ambry-vcr/src/main/java/com/github/ambry/vcr/AmbryCloudRequests.java