private void fetchDataFromReadSet(MessageFormatMetrics metrics) throws MessageFormatException {
    // get size
    int messageCount = readSet.count();
    ArrayList<ByteBuf> dataFromReadSet = new ArrayList<>(messageCount);
    try {
      // for each message, determine the offset and size that needs to be sent based on the flag
      sendInfoList = new ArrayList<>(messageCount);
      messageMetadataList = new ArrayList<>(messageCount);
      logger.trace("Calculate offsets of messages for one partition, MessageFormatFlag : {} number of messages : {}",
          flag, messageCount);
      for (int i = 0; i < messageCount; i++) {
        StoreKey storeKey = readSet.getKeyAt(i);
        if (flag == MessageFormatFlags.All || flag == MessageFormatFlags.Blob) {
          // just copy over the total size and use relative offset to be 0
          // We do not have to check any version in this case as we dont
          // have to read any data to deserialize anything.
          readSet.doPrefetch(i, 0, readSet.sizeInBytes(i));
          dataFromReadSet.add(readSet.getPrefetchedData(i));
          // If we can successfully deserialize BlobAll, then the message content is correct
          try {
            ByteBuf messageData = readSet.getPrefetchedData(i).duplicate();
            MessageFormatRecord.deserializeBlobAll(new NettyByteBufDataInputStream(messageData), storeKeyFactory);
          } catch (Exception e) {
            metrics.messageFormatExceptionCount.inc();
            logger.error(
                "Failed to deserialize the BlobAll from message format send for StoreKey {}. The data source for this key is {}",
                storeKey, readSet.getDataSource(i), e);
          }

          if (flag == MessageFormatFlags.All) {
            sendInfoList.add(i, new SendInfo(0, readSet.sizeInBytes(i)));
            messageMetadataList.add(i, null);
            totalSizeToWrite += readSet.sizeInBytes(i);
          } else if (flag == MessageFormatFlags.Blob) {
            ByteBuf blobAll = readSet.getPrefetchedData(i);
            InputStream blobInputStream = new ByteBufInputStream(blobAll);
            MessageHeader_Format headerFormat = parseHeaderAndVerifyStoreKey(blobInputStream, i);
            MessageMetadata messageMetadata = null;
            if (headerFormat.hasEncryptionKeyRecord()) {
              // If encryption key exists, MessageMetadata with encryption key is needed.
              ByteBuf duplicatedByteBuf = blobAll.duplicate();
              duplicatedByteBuf.setIndex(headerFormat.getBlobEncryptionKeyRecordRelativeOffset(),
                  headerFormat.getBlobEncryptionKeyRecordRelativeOffset()
                      + headerFormat.getBlobEncryptionKeyRecordSize());
              messageMetadata =
                  new MessageMetadata(deserializeBlobEncryptionKey(new ByteBufInputStream(duplicatedByteBuf)));
            }
            messageMetadataList.add(messageMetadata);
            sendInfoList.add(i,
                new SendInfo(headerFormat.getBlobRecordRelativeOffset(), headerFormat.getBlobRecordSize()));
            totalSizeToWrite += headerFormat.getBlobRecordSize();

            // Adjust underlying ByteBuf reader and writer index.
            blobAll.setIndex(headerFormat.getBlobRecordRelativeOffset(),
                (int) (headerFormat.getBlobRecordRelativeOffset() + headerFormat.getBlobRecordSize()));
          }
        } else {
          BufferedInputStream bufferedInputStream =
              new BufferedInputStream(new MessageReadSetIndexInputStream(readSet, i, 0),
                  BUFFERED_INPUT_STREAM_BUFFER_SIZE);

          MessageHeader_Format headerFormat = parseHeaderAndVerifyStoreKey(bufferedInputStream, i);

          long startTime = SystemTime.getInstance().milliseconds();
          if (flag == MessageFormatFlags.BlobProperties) {
            sendInfoList.add(i, new SendInfo(headerFormat.getBlobPropertiesRecordRelativeOffset(),
                headerFormat.getBlobPropertiesRecordSize()));
            messageMetadataList.add(null);
            readSet.doPrefetch(i, headerFormat.getBlobPropertiesRecordRelativeOffset(),
                headerFormat.getBlobPropertiesRecordSize());
            totalSizeToWrite += headerFormat.getBlobPropertiesRecordSize();
            logger.trace("Calculate offsets, get total size of blob properties time: {}",
                SystemTime.getInstance().milliseconds() - startTime);
            logger.trace("Sending blob properties for message relativeOffset : {} size : {}",
                sendInfoList.get(i).relativeOffset(), sendInfoList.get(i).sizetoSend());
          } else if (flag == MessageFormatFlags.BlobUserMetadata) {
            messageMetadataList.add(headerFormat.hasEncryptionKeyRecord() ? new MessageMetadata(
                extractEncryptionKey(i, headerFormat.getBlobEncryptionKeyRecordRelativeOffset(),
                    headerFormat.getBlobEncryptionKeyRecordSize())) : null);
            sendInfoList.add(i, new SendInfo(headerFormat.getUserMetadataRecordRelativeOffset(),
                headerFormat.getUserMetadataRecordSize()));
            readSet.doPrefetch(i, headerFormat.getUserMetadataRecordRelativeOffset(),
                headerFormat.getUserMetadataRecordSize());
            totalSizeToWrite += headerFormat.getUserMetadataRecordSize();
            logger.trace("Calculate offsets, get total size of user metadata time: {}",
                SystemTime.getInstance().milliseconds() - startTime);
            logger.trace("Sending user metadata for message relativeOffset : {} size : {}",
                sendInfoList.get(i).relativeOffset(), sendInfoList.get(i).sizetoSend());
          } else if (flag == MessageFormatFlags.BlobInfo) {
            messageMetadataList.add(headerFormat.hasEncryptionKeyRecord() ? new MessageMetadata(
                extractEncryptionKey(i, headerFormat.getBlobEncryptionKeyRecordRelativeOffset(),
                    headerFormat.getBlobEncryptionKeyRecordSize())) : null);
            sendInfoList.add(i, new SendInfo(headerFormat.getBlobPropertiesRecordRelativeOffset(),
                headerFormat.getBlobPropertiesRecordSize() + headerFormat.getUserMetadataRecordSize()));
            readSet.doPrefetch(i, headerFormat.getBlobPropertiesRecordRelativeOffset(),
                headerFormat.getBlobPropertiesRecordSize() + headerFormat.getUserMetadataRecordSize());
            totalSizeToWrite += headerFormat.getBlobPropertiesRecordSize() + headerFormat.getUserMetadataRecordSize();
            logger.trace("Calculate offsets, get total size of blob info time: {}",
                SystemTime.getInstance().milliseconds() - startTime);
            logger.trace(
                "Sending blob info (blob properties + user metadata) for message relativeOffset : {} " + "size : {}",
                sendInfoList.get(i).relativeOffset(), sendInfoList.get(i).sizetoSend());
          } else {
            throw new MessageFormatException("Unknown flag in request " + flag, MessageFormatErrorCodes.IOError);
          }
          dataFromReadSet.add(readSet.getPrefetchedData(i));
        }
      }
      if (messageCount == 0) {
        messageContent = Unpooled.EMPTY_BUFFER;
      } else if (messageCount == 1) {
        messageContent = dataFromReadSet.get(0);
      } else {
        CompositeByteBuf compositeByteBuf = dataFromReadSet.get(0).alloc().compositeHeapBuffer(messageCount);
        for (ByteBuf data : dataFromReadSet) {
          compositeByteBuf.addComponent(true, data);
        }
        messageContent = compositeByteBuf;
      }
    } catch (Exception e) {
      for (ByteBuf data : dataFromReadSet) {
        data.release();
      }
      if (!(e instanceof MessageFormatException)) {
        logger.error("Error when calculating offsets", e);
        throw new MessageFormatException("IOError when calculating offsets ", e, MessageFormatErrorCodes.IOError);
      } else {
        throw (MessageFormatException) e;
      }
    }
  }
Fetch data from the MessageReadSet that needs to be sent over the network based on the type of data requested as
indicated by the flags
ambry-messageformat/src/main/java/com/github/ambry/messageformat/MessageFormatSend.java