@Override
  public void handleTtlUpdateRequest(NetworkRequest request) throws InterruptedException {

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

    TtlUpdateRequest updateRequest;

    // This is a case where handleTtlUpdateRequest 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.
    updateRequest =
        (TtlUpdateRequest) ((LocalRequestResponseChannel.LocalChannelRequest) request).getRequestInfo().getRequest();

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

    ServerErrorCode error =
        validateRequest(updateRequest.getBlobId().getPartition(), RequestOrResponseType.TtlUpdateRequest, false);
    if (error != ServerErrorCode.NoError) {
      logger.error("Validating TtlUpdateRequest failed with error {} for request {}", error, updateRequest);
      // Send Response
      requestResponseChannel.sendResponse(
          new TtlUpdateResponse(updateRequest.getCorrelationId(), updateRequest.getClientId(), error), request, null);
    } else {
      BlobId blobId = updateRequest.getBlobId();
      MessageInfo info = new MessageInfo.Builder(blobId, -1, blobId.getAccountId(), blobId.getContainerId(),
          updateRequest.getOperationTimeInMs()).isTtlUpdated(true)
          .expirationTimeInMs(updateRequest.getExpiresAtMs())
          .lifeVersion(MessageInfo.LIFE_VERSION_FROM_FRONTEND)
          .build();
      CloudBlobStore cloudBlobStore = (CloudBlobStore) storeManager.getStore(updateRequest.getBlobId().getPartition());
      cloudBlobStore.updateTtlAsync(Collections.singletonList(info)).whenCompleteAsync((unused, throwable) -> {
        if (throwable != null) {
          Exception ex = Utils.extractFutureExceptionCause(throwable);
          if (ex instanceof StoreException) {
            logger.error("Store exception on a TTL update with error code {} for request {}",
                ((StoreException) ex).getErrorCode(), updateRequest, ex);
            response.set(new TtlUpdateResponse(updateRequest.getCorrelationId(), updateRequest.getClientId(),
                ErrorMapping.getStoreErrorMapping(((StoreException) ex).getErrorCode())));
          } else {
            logger.error("Unknown exception for TTL update request{}", updateRequest, ex);
            response.set(new TtlUpdateResponse(updateRequest.getCorrelationId(), updateRequest.getClientId(),
                ServerErrorCode.UnknownError));
          }
        } else {
          response.set(new TtlUpdateResponse(updateRequest.getCorrelationId(), updateRequest.getClientId(),
              ServerErrorCode.NoError));
        }
        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