@Override
    void handleBody(InputStream payload, MessageMetadata messageMetadata, MessageInfo messageInfo)
        throws IOException, MessageFormatException {
      /* If we find blob metadata in frontend cache, validate it and use it. */
      BlobProperties receivedBlobProperties = MessageFormatRecord.deserializeBlobProperties(payload);
      updateTtlIfRequired(receivedBlobProperties, messageInfo);
      /*
       * Some cached variables are mutable.
       * Update them and then compare if other immutable variables are same or not.
       */
      updateTtlIfRequired(serverBlobProperties, messageInfo);
      lifeVersion = messageInfo.getLifeVersion();
      blobInfo.setLifeVersion(lifeVersion);
      if (!serverBlobProperties.equals(receivedBlobProperties)) {
        /*
         * This is a rare error case.
         * If we find blob properties is in our frontend cache, we request just blobInfo from backend to validate
         * the cache entry. If cached blob properties does not match received blob properties, due to a random bit flip for eg.,
         * we must delete the cache entry and throw a retriable exception to the user.
         * The next request will re-fill the cache with a valid entry if the blob stills exists.
         * We do not need to decrypt anything at this point.
         */
        String reason = "Cached blob property does not match received blob property";
        logger.error("[{}] {} for blobId = {}, cached blob property = {}, received blob property = {}",
            blobMetadataCache.getCacheId(), reason, blobId, serverBlobProperties, receivedBlobProperties);
        deleteMetadata(reason);
        onInvalidCacheEntry(new RouterException(reason, RouterErrorCode.UnexpectedInternalError));
        return;
      }
      getOptions().ageAtAccessTracker.trackAgeAtAccess(serverBlobProperties.getCreationTimeInMs());

      /*
       * If we reach here, then the cached metadata is valid and already decrypted from a previous GET/RANGE request.
       * Just use it directly and initiate fetching data chunks.
       */
      if (isRangeRequest() && resolveRangeProper(totalSize)) {
        chunkMetadataList = compositeBlobInfo.getStoreKeysInByteRange(resolvedByteRange.getStartOffset(),
            resolvedByteRange.getEndOffset());
        initializeDataChunks();
      }
    }
Validate cached metadata chunk and proceed to request data chunks
@param payload Bytes received from backend
@param messageMetadata  Metadata associated with messages sent out
@param messageInfo Message info class that contains basic info about a message
@throws IOException
@throws MessageFormatException
ambry-router/src/main/java/com/github/ambry/router/GetBlobOperation.java