private void verify(String dataDir) throws Exception {
    final String Cleanup_Token_Filename = "cleanuptoken";

    FileWriter fileWriter = null;
    try {
      fileWriter = new FileWriter(new File(outFile));
      long offsetInCleanupToken = getOffsetFromCleanupToken(new File(dataDir, Cleanup_Token_Filename));
      rangeMap = new HashMap<BlobId, IndexValue>();
      offRangeMap = new HashMap<BlobId, IndexValue>();
      long lastEligibleSegmentEndOffset = readAndPopulateIndex(offsetInCleanupToken);

      // 2. Scan the log and check against blobMap
      File logFile = new File(dataDir, "log_current");
      RandomAccessFile randomAccessFile = new RandomAccessFile(logFile, "r");
      InputStream streamlog = Channels.newInputStream(randomAccessFile.getChannel());

      long currentOffset = 0;
      System.out.println("Starting scan from offset " + currentOffset + " to " + offsetInCleanupToken);

      long lastOffsetToLookFor = lastEligibleSegmentEndOffset;
      boolean seeking = false;
      while (currentOffset < lastOffsetToLookFor) {
        try {
          short version = randomAccessFile.readShort();
          if (version == 1) {
            seeking = false;
            ByteBuffer buffer = ByteBuffer.allocate(MessageFormatRecord.MessageHeader_Format_V1.getHeaderSize());
            buffer.putShort(version);
            randomAccessFile.read(buffer.array(), 2, buffer.capacity() - 2);
            buffer.rewind();
            MessageFormatRecord.MessageHeader_Format_V1 header =
                new MessageFormatRecord.MessageHeader_Format_V1(buffer);

            // read blob id
            BlobId id;
            id = new BlobId(new DataInputStream(streamlog), map);

            IndexValue indexValue = rangeMap.get(id);
            boolean isDeleted = false;
            if (indexValue == null) {
              throw new IllegalStateException("Key in log not found in index " + id);
            } else if (indexValue.isFlagSet(IndexValue.Flags.Delete_Index)) {
              isDeleted = true;
            }

            if (header.getBlobPropertiesRecordRelativeOffset()
                != MessageFormatRecord.Message_Header_Invalid_Relative_Offset) {
              BlobProperties props;
              ByteBuffer metadata;
              BlobData output;
              try {
                props = MessageFormatRecord.deserializeBlobProperties(streamlog);
                metadata = MessageFormatRecord.deserializeUserMetadata(streamlog);
                output = MessageFormatRecord.deserializeBlob(streamlog);
              } catch (MessageFormatException e) {
                if (!isDeleted) {
                  corruptNonDeleted++;
                } else {
                  corruptDeleted++;
                }
                throw e;
              }

              if (isDeleted) {
                ByteBuf byteBuf = output.content();
                try {
                  if (!verifyZeroed(metadata.array()) || !verifyZeroed(
                      Utils.readBytesFromByteBuf(byteBuf, new byte[(int) output.getSize()], 0,
                          (int) output.getSize()))) {
                    /* If the offset in the index is different from that in the log, hard delete wouldn't have been
                       possible and we just saw a duplicate put for the same key, otherwise we missed a hard delete. */
                    if (currentOffset == indexValue.getOriginalMessageOffset()) {
                      notHardDeletedErrorCount++;
                    } else {
                      // the assumption here is that this put has been lost as far as the index is concerned due to
                      // a duplicate put. Of course, these shouldn't happen anymore, we are accounting for past
                      // bugs.
                      duplicatePuts++;
                    }
                  } else {
                    hardDeletedPuts++;
                  }
                } finally {
                  byteBuf.release();
                }
              } else {
                unDeletedPuts++;
              }
            } else if (MessageFormatRecord.deserializeUpdateRecord(streamlog).getType().equals(SubRecord.Type.DELETE)) {
              deletes++;
            }
            currentOffset += (header.getMessageSize() + buffer.capacity() + id.sizeInBytes());
          } else {
            throw new IllegalStateException("Unknown version for entry");
          }
        } catch (MessageFormatException e) {
          if (!seeking) {
            invalidEntriesInlog = true;
            e.printStackTrace();
            seeking = true;
          }
          randomAccessFile.seek(++currentOffset);
        } catch (IOException e) {
          if (!seeking) {
            invalidEntriesInlog = true;
            e.printStackTrace();
            seeking = true;
          }
          randomAccessFile.seek(++currentOffset);
        } catch (IllegalArgumentException e) {
          if (!seeking) {
            invalidEntriesInlog = true;
            e.printStackTrace();
            seeking = true;
          }
          randomAccessFile.seek(++currentOffset);
        } catch (IllegalStateException e) {
          if (!seeking) {
            invalidEntriesInlog = true;
            e.printStackTrace();
            seeking = true;
          }
          randomAccessFile.seek(++currentOffset);
        } catch (Exception e) {
          e.printStackTrace(System.err);
          invalidEntriesInlog = true;
          randomAccessFile.seek(++currentOffset);
          break;
        }
      }
      String msg = ("\n============");
      msg += "\ninvalidEntriesInlog? " + (invalidEntriesInlog ? "Yes" : "No");
      msg += "\nnotHardDeletedErrorCount: " + notHardDeletedErrorCount;
      msg += "\ncorruptNonDeletedCount:" + corruptNonDeleted;
      msg += "\n========";
      msg += "\ncorruptDeleted:" + corruptDeleted;
      msg += "\nduplicatePuts: " + duplicatePuts;
      msg += "\nundeleted Put Records: " + unDeletedPuts;
      msg += "\nhard deleted Put Records: " + hardDeletedPuts;
      msg += "\nDelete Records: " + deletes;
      msg += "\n============";
      fileWriter.write(msg);
      System.out.println(msg);
    } finally {
      if (fileWriter != null) {
        fileWriter.flush();
        fileWriter.close();
      }
    }
  }
This method ensures that all the blobs that should have been hard deleted are indeed hard deleted and the rest
are untouched. Optionally, if the state of the dataDir prior to hard deletes being enabled is available, then
compares the non-hard deleted records between the two and ensure they are exactly the same.

Here's the algorithm:

1. Reads cleanupToken and gets the conservative offset till which hard deletes surely are complete.
2. Reads the index files and stores everything upto the conservative offset above in a "rangeMap" and the later
   entries in an "offRangeMap".
3. Goes through the log file upto the conservative offset, and for each entry checks whether deserialization
   happens successfully, whether it is zeroed out if it is deleted, whether the content is the same as in the
   older data file (if there is one) if it undeleted, whether duplicate puts were encountered etc.

@throws IOException
ambry-tools/src/main/java/com/github/ambry/store/HardDeleteVerifier.java