@Override
    void handleBody(InputStream payload, MessageMetadata messageMetadata, MessageInfo messageInfo)
        throws IOException, MessageFormatException {
      if (!successfullyDeserialized) {
        BlobData blobData;
        ByteBuffer encryptionKey;
        byte[] userMetadata = null;

        ByteBuffer rawPayloadBuffer = null;
        boolean rawMode = options.getBlobOptions.isRawMode();
        if (rawMode) {
          if (messageInfo.getSize() > Integer.MAX_VALUE) {
            throw new IllegalStateException("Payload too large for blob " + blobId.getID());
          }
          // TODO: avoid extra copy using InputStreamReadableStreamChannel
          byte[] payloadBytes = new byte[(int) messageInfo.getSize()];
          payload.read(payloadBytes);
          payload = new ByteArrayInputStream(payloadBytes);
          rawPayloadBuffer = ByteBuffer.wrap(payloadBytes);
        }

        if (getOperationFlag() == MessageFormatFlags.Blob) {
          blobData = MessageFormatRecord.deserializeBlob(payload);
          encryptionKey = messageMetadata == null ? null : messageMetadata.getEncryptionKey();
        } else {
          BlobAll blobAll = MessageFormatRecord.deserializeBlobAll(payload, blobIdFactory);
          BlobInfo serverBlobInfo = blobAll.getBlobInfo();
          updateTtlIfRequired(serverBlobInfo.getBlobProperties(), messageInfo);
          getOptions().ageAtAccessTracker.trackAgeAtAccess(serverBlobInfo.getBlobProperties().getCreationTimeInMs());
          blobData = blobAll.getBlobData();
          encryptionKey = blobAll.getBlobEncryptionKey();
          if (Objects.isNull(encryptionKey) || options.getChunkIdsOnly) {
            // set blobInfo iff:
            //   1. blob is not encrypted OR,
            //   2. getChunkIdsOnly is set.
            // If the blob is encrypted, then the user metadata is encrypted too. So in that case,
            // maybeProcessCallbackAndComplete() will set the appropriate value for blobInfo after decryption.
            // In the case of getChunkIdsOnly call, we don't need the user metadata but still need blob
            // properties to evaluate if the metadata chunk is already updated for update requests. So we allow blobInfo
            // to be set here.
            blobInfo = serverBlobInfo;
            blobInfo.setLifeVersion(messageInfo.getLifeVersion());
          } else {
            serverBlobProperties = serverBlobInfo.getBlobProperties();
            lifeVersion = messageInfo.getLifeVersion();
            userMetadata = serverBlobInfo.getUserMetadata();
          }
        }
        blobType = blobData.getBlobType();
        isChunkCompressed = blobData.isCompressed();
        logger.trace(
            "In FirstGetChunk.handleBody(), is Chunk compressed = " + isChunkCompressed + " for blob ID = " + blobId);
        chunkIndexToBuf = new ConcurrentHashMap<>();
        chunkIndexToBufWaitingForRelease = new ConcurrentHashMap<>();
        if (rawMode) {
          if (blobData != null) {
            // RawMode, release blob data.
            ReferenceCountUtil.safeRelease(blobData);
          }
          // Return the raw bytes from storage
          if (encryptionKey != null) {
            chunkIdIterator = null;
            dataChunks = null;
            chunkIndex = 0;
            numChunksTotal = 1;
            ByteBuf decompressedBuffer = decompressContent(Unpooled.wrappedBuffer(rawPayloadBuffer));
            if (decompressedBuffer != null) {
              chunkIndexToBuf.put(0, decompressedBuffer);
              numChunksRetrieved.set(1);
            }
          } else {
            setOperationException(new IllegalStateException("Only encrypted blobs supported in raw mode"));
          }
        } else {
          if (blobType == BlobType.MetadataBlob) {
            handleMetadataBlob(blobData, userMetadata, encryptionKey);
          } else {
            handleSimpleBlob(blobData, userMetadata, encryptionKey);
          }
        }
        successfullyDeserialized = true;
      } else {
        // Currently, regardless of the successTarget, only the first successful response is honored. Subsequent ones
        // are ignored. If ever in the future, we need some kind of reconciliation, this is the place
        // to do that. (Only after the reconciliation will the state be marked as complete).
      }
    }
{@inheritDoc}
<br>
It would help to keep in mind while going through this method that the first chunk is either a metadata chunk
or the only chunk of the blob.
ambry-router/src/main/java/com/github/ambry/router/GetBlobOperation.java