private void compareIndexEntriesToLogContent(File indexFile, boolean checkLogEndOffsetMatch) throws Exception {
    if (!indexFile.exists()) {
      throw new IllegalArgumentException("File does not exist " + indexFile);
    }
    final Timer.Context context = metrics.compareIndexFileToLogTimeMs.time();
    try {
      logger.info("Dumping index {}", indexFile.getAbsolutePath());
      StoreKeyFactory storeKeyFactory = new BlobIdFactory(clusterMap);
      StoreConfig config = new StoreConfig(new VerifiableProperties(new Properties()));
      MetricRegistry metricRegistry = new MetricRegistry();
      StoreMetrics storeMetrics = new StoreMetrics(metricRegistry);
      IndexSegment segment = new IndexSegment(indexFile, false, storeKeyFactory, config, storeMetrics,
          new Journal(indexFile.getParent(), 0, 0), time);
      Offset startOffset = segment.getStartOffset();
      TreeMap<Long, Long> coveredRanges = new TreeMap<>();
      String logFileName = segment.getLogSegmentName().toFilename();
      File logFile = new File(indexFile.getParent(), logFileName);
      if (!logFile.exists()) {
        throw new IllegalStateException("Log file does not exist " + logFile);
      }
      RandomAccessFile randomAccessFile = new RandomAccessFile(logFile, "r");
      long logFileSize = randomAccessFile.getChannel().size();
      List<MessageInfo> entries = new ArrayList<>();
      segment.getEntriesSince(null, new FindEntriesCondition(Long.MAX_VALUE), entries, new AtomicLong(0), false);
      for (MessageInfo entry : entries) {
        StoreKey key = entry.getStoreKey();
        IndexValue value = segment.find(key).last();
        boolean isDeleted = value.isFlagSet(IndexValue.Flags.Delete_Index);
        if (value.getOffset().getOffset() < logFileSize) {
          boolean success = readFromLogAndVerify(randomAccessFile, key.getID(), value, coveredRanges);
          if (success) {
            if (isDeleted) {
              long originalOffset = value.getOriginalMessageOffset();
              if (originalOffset != -1) {
                if (!coveredRanges.containsKey(originalOffset)) {
                  if (startOffset.getOffset() > originalOffset) {
                    logger.trace(
                        "Put Record at {} with delete msg offset {} ignored because it is prior to startOffset {}",
                        originalOffset, value.getOffset(), startOffset);
                  } else {
                    try {
                      DumpDataHelper.LogBlobRecordInfo logBlobRecordInfo =
                          DumpDataHelper.readSingleRecordFromLog(randomAccessFile, originalOffset, clusterMap,
                              currentTimeInMs, metrics);
                      coveredRanges.put(originalOffset, originalOffset + logBlobRecordInfo.totalRecordSize);
                      logger.trace(
                          "PUT Record {} with start offset {} and end offset {} for a delete msg {} at offset {} ",
                          logBlobRecordInfo.blobId, originalOffset,
                          (originalOffset + logBlobRecordInfo.totalRecordSize), key.getID(), value.getOffset());
                      if (!logBlobRecordInfo.blobId.getID().equals(key.getID())) {
                        logger.error("BlobId value mismatch between delete record {} and put record {}", key.getID(),
                            logBlobRecordInfo.blobId.getID());
                      }
                    } catch (IllegalArgumentException e) {
                      metrics.logDeserializationError.inc();
                      logger.error(
                          "Illegal arg exception thrown at  {}, while reading blob starting at offset {} with exception: ",
                          randomAccessFile.getChannel().position(), originalOffset, e);
                    } catch (MessageFormatException e) {
                      metrics.logDeserializationError.inc();
                      logger.error(
                          "MessageFormat exception thrown at  {} while reading blob starting at offset {} with exception: ",
                          randomAccessFile.getChannel().position(), originalOffset, e);
                    } catch (EOFException e) {
                      metrics.endOfFileOnDumpLogError.inc();
                      logger.error("EOFException thrown at {} ", randomAccessFile.getChannel().position(), e);
                    } catch (Exception e) {
                      metrics.unknownErrorOnDumpIndex.inc();
                      logger.error("Unknown exception thrown {} ", e.getMessage(), e);
                    }
                  }
                }
              }
            }
          } else {
            metrics.indexToLogBlobRecordComparisonFailure.inc();
            logger.error("Failed for key {} with value {} ", key, value);
          }
        } else {
          logger.trace("Blob's {} offset {} is outside of log size {}, with a diff of {}", key,
              value.getOffset().getOffset(), logFileSize, (value.getOffset().getOffset() - logFileSize));
        }
      }
      throttler.maybeThrottle(entries.size());
      long indexEndOffset = segment.getEndOffset().getOffset();
      if (checkLogEndOffsetMatch && indexEndOffset != randomAccessFile.length()) {
        metrics.indexLogEndOffsetMisMatchError.inc();
        logger.error("Log end offset {} and index end offset {} do not match", randomAccessFile.length(),
            indexEndOffset);
      }
      logRangesNotCovered(coveredRanges, indexEndOffset);
    } finally {
      context.stop();
    }
  }
Compares every entry in an index file with those in the log. Checks to see if each blob in index is successfully
deserializable from the log
@param indexFile the file that represents the index segment.
@param checkLogEndOffsetMatch if {@code true}, checks that the end offset of the log matches the end offset of the
                              index.
@throws Exception
ambry-tools/src/main/java/com/github/ambry/store/DumpDataTool.java