public IndexProcessingResults processIndex(File replicaDir, Set<StoreKey> filterSet, long currentTimeMs,
      boolean detectDuplicatesAcrossKeys) throws Exception {
    verifyPath(replicaDir, true);
    final Timer.Context context = metrics.dumpReplicaIndexesTimeMs.time();
    try {
      File[] segmentFiles = getSegmentFilesFromDir(replicaDir);
      SortedMap<File, List<IndexEntry>> segmentFileToIndexEntries = getAllEntriesFromIndex(segmentFiles);
      Map<StoreKey, Info> keyToState = new HashMap<>();
      long processedCount = 0;
      long putCount = 0;
      long deleteCount = 0;
      long ttlUpdateCount = 0;
      long craftedIdCount = 0;
      Set<StoreKey> duplicatePuts = new HashSet<>();
      Set<StoreKey> putAfterUpdates = new HashSet<>();
      Set<StoreKey> duplicateDeletes = new HashSet<>();
      Set<StoreKey> duplicateUpdates = new HashSet<>();
      Set<StoreKey> updatesAfterDeletes = new HashSet<>();
      long indexSegmentCount = segmentFileToIndexEntries.size();
      for (int i = 0; i < indexSegmentCount; i++) {
        List<IndexEntry> entries = segmentFileToIndexEntries.get(segmentFiles[i]);
        boolean isInRecentIndexSegment = i >= indexSegmentCount - 2;
        processedCount += entries.size();
        for (IndexEntry entry : entries) {
          StoreKey key = entry.getKey();
          if (filterSet == null || filterSet.isEmpty() || filterSet.contains(key)) {
            if (key instanceof BlobId && BlobId.isCrafted(key.getID())) {
              craftedIdCount++;
            }
            boolean isDelete = entry.getValue().isFlagSet(IndexValue.Flags.Delete_Index);
            boolean isTtlUpdate = !isDelete && entry.getValue().isFlagSet(IndexValue.Flags.Ttl_Update_Index);
            boolean isExpired = DumpDataHelper.isExpired(entry.getValue().getExpiresAtMs(), currentTimeMs);
            boolean isPermanent = entry.getValue().getExpiresAtMs() == Utils.Infinite_Time;
            EnumSet<BlobState> states = isExpired ? EnumSet.of(BlobState.Expired) : EnumSet.noneOf(BlobState.class);
            if (isDelete) {
              deleteCount++;
              states.add(BlobState.Deleted);
            } else if (isTtlUpdate) {
              ttlUpdateCount++;
            } else {
              putCount++;
              if (!isExpired) {
                states.add(BlobState.Valid);
              }
            }
            Info info = keyToState.get(key);
            if (info == null) {
              info = new Info(states, isInRecentIndexSegment, isPermanent);
              if (isTtlUpdate) {
                info.markTtlUpdateSeen();
              }
              keyToState.put(key, info);
            } else if (info.states.contains(BlobState.Deleted)) {
              if (isDelete) {
                duplicateDeletes.add(key);
              } else if (isTtlUpdate) {
                updatesAfterDeletes.add(key);
              } else {
                putAfterUpdates.add(key);
              }
            } else {
              if (isDelete) {
                Info newInfo = new Info(states, isInRecentIndexSegment, isPermanent);
                if (info.isTtlUpdateSeen()) {
                  newInfo.markTtlUpdateSeen();
                }
                keyToState.put(key, newInfo);
              } else if (info.isTtlUpdateSeen()) {
                if (isTtlUpdate) {
                  duplicateUpdates.add(key);
                } else {
                  putAfterUpdates.add(key);
                }
              } else if (isTtlUpdate) {
                states.add(isExpired ? BlobState.Expired : BlobState.Valid);
                Info newInfo = new Info(states, isInRecentIndexSegment, isPermanent);
                newInfo.markTtlUpdateSeen();
                keyToState.put(key, newInfo);
              } else {
                duplicatePuts.add(key);
              }
            }
          }
        }
      }
      Map<StoreKey, StoreKey> duplicateKeys = new HashMap<>();
      if (detectDuplicatesAcrossKeys) {
        logger.info("Converting {} store keys...", keyToState.size());
        Map<StoreKey, StoreKey> conversions = storeKeyConverter.convert(keyToState.keySet());
        logger.info("Store keys converted!");
        conversions.forEach((k, v) -> {
          if (!k.equals(v) && keyToState.containsKey(v)) {
            // if k == v, the processing above should have caught it
            // if v exists in keyToState, it means both k and v exist in the store and this is a problem
            duplicateKeys.put(k, v);
          }
        });
      }
      return new IndexProcessingResults(keyToState, processedCount, putCount, ttlUpdateCount, deleteCount,
          craftedIdCount, duplicatePuts, putAfterUpdates, duplicateDeletes, duplicateUpdates, updatesAfterDeletes,
          duplicateKeys);
    } finally {
      context.stop();
    }
  }
Returns {@link IndexProcessingResults} for each replicaDir in {@code replicaDirs}
@param replicaDirs the directories to examine
@param filterSet the filter set of keys to examine. Can be {@code null}
@param parallelism the desired parallelism. The minimum of this and the size of {@code replicaDirs} is the actual
                   chosen parallelism.
@param detectDuplicatesAcrossKeys if {@code true}, uses the {@link StoreKeyConverter} to detect duplicates across
                                  keys. If set to {@code true}, parallelism has to be 1.
@return @link IndexProcessingResults} for each replicaDir in {@code replicaDirs}
ambry-tools/src/main/java/com/github/ambry/store/DumpIndexTool.java