private HardDeleteInfo getHardDeleteInfo(int readSetIndex) {

    HardDeleteInfo hardDeleteInfo = null;

    try {
      /* Read the version field in the header */
      ByteBuffer headerVersionBuf = ByteBuffer.allocate(Version_Field_Size_In_Bytes);
      readSet.writeTo(readSetIndex, Channels.newChannel(new ByteBufferOutputStream(headerVersionBuf)), 0,
          Version_Field_Size_In_Bytes);
      headerVersionBuf.flip();
      short headerVersion = headerVersionBuf.getShort();
      if (!isValidHeaderVersion(headerVersion)) {
        throw new MessageFormatException(
            "Unknown header version during hard delete " + headerVersion + " storeKey " + readSet.getKeyAt(
                readSetIndex), MessageFormatErrorCodes.UnknownFormatVersion);
      }
      ByteBuffer header = ByteBuffer.allocate(getHeaderSizeForVersion(headerVersion));
      /* Read the rest of the header */
      header.putShort(headerVersion);
      readSet.writeTo(readSetIndex, Channels.newChannel(new ByteBufferOutputStream(header)),
          Version_Field_Size_In_Bytes, header.capacity() - Version_Field_Size_In_Bytes);
      header.flip();
      MessageHeader_Format headerFormat = getMessageHeader(headerVersion, header);
      headerFormat.verifyHeader();
      StoreKey storeKey = storeKeyFactory.getStoreKey(
          new DataInputStream(new MessageReadSetIndexInputStream(readSet, readSetIndex, header.capacity())));
      if (storeKey.compareTo(readSet.getKeyAt(readSetIndex)) != 0) {
        throw new MessageFormatException(
            "Id mismatch between metadata and store - metadataId " + readSet.getKeyAt(readSetIndex) + " storeId "
                + storeKey, MessageFormatErrorCodes.StoreKeyIdMisMatch);
      }

      if (!headerFormat.isPutRecord()) {
        throw new MessageFormatException("Cleanup operation for a non-PUT record is unsupported",
            MessageFormatErrorCodes.IOError);
      } else {
        HardDeleteRecoveryMetadata hardDeleteRecoveryMetadata = recoveryInfoMap.get(storeKey);

        int userMetadataRelativeOffset = headerFormat.getUserMetadataRecordRelativeOffset();
        short userMetadataVersion;
        int userMetadataSize;
        short blobRecordVersion;
        BlobType blobType;
        long blobStreamSize;
        DeserializedUserMetadata userMetadataInfo;
        DeserializedBlob blobRecordInfo;

        if (hardDeleteRecoveryMetadata == null) {
          userMetadataInfo =
              getUserMetadataInfo(readSet, readSetIndex, headerFormat.getUserMetadataRecordRelativeOffset(),
                  headerFormat.getUserMetadataRecordSize());
          userMetadataSize = userMetadataInfo.getUserMetadata().capacity();
          userMetadataVersion = userMetadataInfo.getVersion();

          blobRecordInfo = getBlobRecordInfo(readSet, readSetIndex, headerFormat.getBlobRecordRelativeOffset(),
              headerFormat.getBlobRecordSize());
          blobStreamSize = blobRecordInfo.getBlobData().getSize();
          blobRecordVersion = blobRecordInfo.getVersion();
          blobType = blobRecordInfo.getBlobData().getBlobType();
          hardDeleteRecoveryMetadata =
              new HardDeleteRecoveryMetadata(headerVersion, userMetadataVersion, userMetadataSize, blobRecordVersion,
                  blobType, blobStreamSize, storeKey);
        } else {
          logger.trace("Skipping crc check for user metadata and blob stream fields for key {}", storeKey);
          userMetadataVersion = hardDeleteRecoveryMetadata.getUserMetadataVersion();
          blobRecordVersion = hardDeleteRecoveryMetadata.getBlobRecordVersion();
          blobType = hardDeleteRecoveryMetadata.getBlobType();
          userMetadataSize = hardDeleteRecoveryMetadata.getUserMetadataSize();
          blobStreamSize = hardDeleteRecoveryMetadata.getBlobStreamSize();
        }

        HardDeleteMessageFormatInputStream hardDeleteStream =
            new HardDeleteMessageFormatInputStream(userMetadataRelativeOffset, userMetadataVersion, userMetadataSize,
                blobRecordVersion, blobType, blobStreamSize);

        hardDeleteInfo = new HardDeleteInfo(Channels.newChannel(hardDeleteStream), hardDeleteStream.getSize(),
            hardDeleteStream.getHardDeleteStreamRelativeOffset(), hardDeleteRecoveryMetadata.toBytes());
      }
    } catch (Exception e) {
      logger.error("Exception when reading blob: ", e);
    }
    return hardDeleteInfo;
  }
For the message at readSetIndex, does the following:
   1. Reads the whole blob and does a crc check. If the crc check fails, returns null - this means that the record
   is not retrievable anyway.
   2. Adds to a hard delete replacement write set.
   3. Returns the hard delete info.
ambry-messageformat/src/main/java/com/github/ambry/messageformat/BlobStoreHardDelete.java