private void recover(MessageStoreRecovery recovery) throws StoreException, IOException {
    final Timer.Context context = metrics.recoveryTime.time();
    LogSegment firstSegment = log.getFirstSegment();
    Offset recoveryStartOffset = new Offset(firstSegment.getName(), firstSegment.getStartOffset());
    if (validIndexSegments.size() > 0) {
      IndexSegment indexSegment = validIndexSegments.lastEntry().getValue();
      recoveryStartOffset =
          indexSegment.getEndOffset() == null ? indexSegment.getStartOffset() : indexSegment.getEndOffset();
    }
    boolean recoveryOccurred = false;
    LogSegment logSegmentToRecover = log.getSegment(recoveryStartOffset.getName());
    Set<StoreKey> deleteExpectedKeys = new HashSet<>();
    while (logSegmentToRecover != null) {
      long endOffset = logSegmentToRecover.sizeInBytes();
      logger.info("Index : {} performing recovery on index with start offset {} and end offset {}", dataDir,
          recoveryStartOffset, endOffset);
      MessageStoreRecovery.RecoveryResult recoveryResult =
          recovery.recover(logSegmentToRecover, recoveryStartOffset.getOffset(), endOffset, factory);
      maybePerformaPartialRecovery(logSegmentToRecover, recoveryResult);
      List<MessageInfo> messagesRecovered = recoveryResult.recovered;
      recoveryOccurred = recoveryOccurred || messagesRecovered.size() > 0;
      Offset runningOffset = recoveryStartOffset;
      // Iterate through the recovered messages and update the index
      for (MessageInfo info : messagesRecovered) {
        logger.trace("Index : {} recovering key {} offset {} size {}", dataDir, info.getStoreKey(), runningOffset,
            info.getSize());
        Offset infoEndOffset = new Offset(runningOffset.getName(), runningOffset.getOffset() + info.getSize());
        IndexValue value = findKey(info.getStoreKey());
        if (info.isDeleted()) {
          markAsDeleted(info.getStoreKey(), new FileSpan(runningOffset, infoEndOffset), info, info.getOperationTimeMs(),
              info.getLifeVersion());
          logger.info(
              "Index : {} updated message with key {} by inserting delete entry of size {} ttl {} lifeVersion{}",
              dataDir, info.getStoreKey(), info.getSize(), info.getExpirationTimeInMs(), info.getLifeVersion());
          // removes from the tracking structure if a delete was being expected for the key
          deleteExpectedKeys.remove(info.getStoreKey());
        } else if (info.isUndeleted()) {
          markAsUndeleted(info.getStoreKey(), new FileSpan(runningOffset, infoEndOffset), info,
              info.getOperationTimeMs(), info.getLifeVersion());
          logger.info(
              "Index : {} updated message with key {} by inserting undelete entry of size {} ttl {} lifeVersion {}",
              dataDir, info.getStoreKey(), info.getSize(), info.getExpirationTimeInMs(), info.getLifeVersion());
          if (value == null) {
            // This undelete was forced even though there was no equivalent PUT record - this means that we MUST see
            // a DELETE for this key (because the PUT record is gone, compaction must have cleaned it up because a
            // DELETE must have been present). This also works when there are multiple DELETE and UNDELETE for the same
            // key. For instance, if we have UNDELETE, DELETE and then UNDELETE. The first UNDELETE would expect a DELETE
            // to follow given that the PUT is gone. Then recovery would encounter the first DELETE and remove the key
            // from the expected list. But the second UNDELETE would once again expect to see a DELETE follow. If there
            // is not DELETE after second UNDELETE, an exception would be thrown.
            deleteExpectedKeys.add(info.getStoreKey());
          }
        } else if (info.isTtlUpdated()) {
          markAsPermanent(info.getStoreKey(), new FileSpan(runningOffset, infoEndOffset), info,
              info.getOperationTimeMs(), info.getLifeVersion());
          logger.info(
              "Index : {} updated message with key {} by inserting TTL update entry of size {} ttl {} lifeVersion {}",
              dataDir, info.getStoreKey(), info.getSize(), info.getExpirationTimeInMs(), info.getLifeVersion());
          if (value == null) {
            // this TTL update was forced even though there was no equivalent PUT record - this means that we MUST see
            // a DELETE for this key (because the PUT record is gone, compaction must have cleaned it up because a
            // DELETE must have been present)
            deleteExpectedKeys.add(info.getStoreKey());
          }
        } else if (value != null) {
          throw new StoreException(
              "Illegal message state during recovery. Duplicate PUT record for " + info.getStoreKey(),
              StoreErrorCodes.IndexRecoveryError);
        } else {
          // create a new entry in the index
          IndexValue newValue = new IndexValue(info.getSize(), runningOffset, IndexValue.FLAGS_DEFAULT_VALUE,
              info.getExpirationTimeInMs(), info.getOperationTimeMs(), info.getAccountId(), info.getContainerId(),
              info.getLifeVersion());
          addToIndex(new IndexEntry(info.getStoreKey(), newValue, null), new FileSpan(runningOffset, infoEndOffset));
          logger.info("Index : {} adding new message to index with key {} size {} ttl {} deleted {}", dataDir,
              info.getStoreKey(), info.getSize(), info.getExpirationTimeInMs(), info.isDeleted());
        }
        runningOffset = infoEndOffset;
      }
      logSegmentToRecover = log.getNextSegment(logSegmentToRecover);
      if (logSegmentToRecover != null) {
        recoveryStartOffset = new Offset(logSegmentToRecover.getName(), logSegmentToRecover.getStartOffset());
      }
    }
    if (deleteExpectedKeys.size() > 0) {
      throw new StoreException("Deletes were expected for some keys but were not encountered: " + deleteExpectedKeys,
          StoreErrorCodes.IndexRecoveryError);
    }
    if (recoveryOccurred) {
      metrics.nonzeroMessageRecovery.inc();
    }
    context.stop();
  }
Recovers a segment given the end offset in the log and a recovery handler
<p/>
Assumed to run only on startup.
@param recovery The recovery handler that is used to perform the recovery
@throws StoreException
@throws IOException
ambry-store/src/main/java/com/github/ambry/store/PersistentIndex.java