@Override
  public void handleUndeleteRequest(NetworkRequest request) throws IOException, InterruptedException {
    UndeleteRequest undeleteRequest;
    if (request instanceof LocalChannelRequest) {
      // 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) ((LocalChannelRequest) request).getRequestInfo().getRequest();
    } else {
      undeleteRequest = UndeleteRequest.readFrom(new DataInputStream(request.getInputStream()), clusterMap);
    }
    long requestQueueTime = SystemTime.getInstance().milliseconds() - request.getStartTimeInMs();
    long totalTimeSpent = requestQueueTime;
    long startTime = SystemTime.getInstance().milliseconds();
    UndeleteResponse response = null;
    Store storeToUndelete;
    StoreKey convertedStoreKey;
    try {
      convertedStoreKey = getConvertedStoreKeys(Collections.singletonList(undeleteRequest.getBlobId())).get(0);
      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 = new UndeleteResponse(undeleteRequest.getCorrelationId(), undeleteRequest.getClientId(), error);
      } else {
        BlobId convertedBlobId = (BlobId) convertedStoreKey;
        MessageInfo info = new MessageInfo.Builder(convertedBlobId, -1, convertedBlobId.getAccountId(),
            convertedBlobId.getContainerId(), undeleteRequest.getOperationTimeMs()).isUndeleted(true)
            .lifeVersion(MessageInfo.LIFE_VERSION_FROM_FRONTEND)
            .build();
        storeToUndelete = storeManager.getStore(undeleteRequest.getBlobId().getPartition());
        short lifeVersion = storeToUndelete.undelete(info);
        response = new UndeleteResponse(undeleteRequest.getCorrelationId(), undeleteRequest.getClientId(), lifeVersion);
        if (notification != null) {
          notification.onBlobReplicaUndeleted(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.TTLExpired) {
        metrics.ttlExpiredError.inc();
      } else if (e.getErrorCode() == StoreErrorCodes.IDDeletedPermanently) {
        metrics.idDeletedError.inc();
      } else if (e.getErrorCode() == StoreErrorCodes.LifeVersionConflict) {
        metrics.lifeVersionConflictError.inc();
      } else if (e.getErrorCode() == StoreErrorCodes.IDNotDeleted) {
        metrics.idNotDeletedError.inc();
      } else if (e.getErrorCode() == StoreErrorCodes.IDUndeleted) {
        metrics.idUndeletedError.inc();
      } else if (e.getErrorCode() == StoreErrorCodes.AuthorizationFailure) {
        metrics.undeleteAuthorizationFailure.inc();
      } else {
        logInErrorLevel = true;
        metrics.unExpectedStoreUndeleteError.inc();
      }
      if (logInErrorLevel) {
        logger.error("Store exception on a undelete with error code {} for request {}", e.getErrorCode(),
            undeleteRequest, e);
      } else {
        logger.trace("Store exception on a undelete with error code {} for request {}", e.getErrorCode(),
            undeleteRequest, e);
      }
      if (e.getErrorCode() == StoreErrorCodes.IDUndeleted) {
        if (e instanceof IdUndeletedStoreException) {
          response = new UndeleteResponse(undeleteRequest.getCorrelationId(), undeleteRequest.getClientId(),
              ((IdUndeletedStoreException) e).getLifeVersion(), ServerErrorCode.BlobAlreadyUndeleted);
        } else {
          response = new UndeleteResponse(undeleteRequest.getCorrelationId(), undeleteRequest.getClientId(),
              MessageInfo.LIFE_VERSION_FROM_FRONTEND, ServerErrorCode.BlobAlreadyUndeleted);
        }
      } else {
        response = new UndeleteResponse(undeleteRequest.getCorrelationId(), undeleteRequest.getClientId(),
            ErrorMapping.getStoreErrorMapping(e.getErrorCode()));
      }
    } catch (Exception e) {
      logger.error("Unknown exception for undelete request {}", undeleteRequest, e);
      response = new UndeleteResponse(undeleteRequest.getCorrelationId(), undeleteRequest.getClientId(),
          ServerErrorCode.UnknownError);
      metrics.unExpectedStoreUndeleteError.inc();
    } finally {
      long processingTime = SystemTime.getInstance().milliseconds() - startTime;
      totalTimeSpent += processingTime;
      publicAccessLogger.info("{} {} processingTime {}", undeleteRequest, response, processingTime);
      // Update request metrics.
      RequestMetricsUpdater metricsUpdater = new RequestMetricsUpdater(requestQueueTime, processingTime, 0, 0, false);
      undeleteRequest.accept(metricsUpdater);
    }
    requestResponseChannel.sendResponse(response, request,
        new ServerNetworkResponseMetrics(metrics.undeleteBlobResponseQueueTimeInMs, metrics.undeleteBlobSendTimeInMs,
            metrics.undeleteBlobTotalTimeInMs, null, null, totalTimeSpent));
  }
Get the Blob from the remote replica
@param replicateBlobRequest the {@link ReplicateBlobRequest}
@return a pair of the {@link ServerErrorCode} and the {@link GetResponse}.
ambry-server/src/main/java/com/github/ambry/server/AmbryRequests.java