FindInfo findDeletedEntriesSince(FindToken token, long maxTotalSizeOfEntries, long endTimeSeconds)
      throws StoreException {
    StoreFindToken storeToken = (StoreFindToken) token;
    StoreFindToken newToken;
    List<MessageInfo> messageEntries = new ArrayList<MessageInfo>();

    if (storeToken.getType().equals(FindTokenType.IndexBased)) {
      // Case 1: index based
      // Find the index segment corresponding to the token indexStartOffset.
      // Get entries starting from the token Key in this index.
      newToken = findEntriesFromSegmentStartOffset(storeToken.getOffset(), storeToken.getStoreKey(), messageEntries,
          new FindEntriesCondition(maxTotalSizeOfEntries, endTimeSeconds), validIndexSegments, false);
      if (newToken.getType().equals(FindTokenType.Uninitialized)) {
        newToken = storeToken;
      }
    } else {
      // journal based or empty
      Offset offsetToStart = storeToken.getOffset();
      boolean inclusive = false;
      if (storeToken.getType().equals(FindTokenType.Uninitialized)) {
        offsetToStart = getStartOffset();
        inclusive = true;
      }
      List<JournalEntry> entries = journal.getEntriesSince(offsetToStart, inclusive);
      // we deliberately obtain a snapshot of the index segments AFTER fetching from the journal. This ensures that
      // any and all entries returned from the journal are guaranteed to be in the obtained snapshot of indexSegments.
      ConcurrentSkipListMap<Offset, IndexSegment> indexSegments = validIndexSegments;

      Offset offsetEnd = offsetToStart;
      if (entries != null) {
        // Case 2: offset based, and offset still in journal
        IndexSegment currentSegment = indexSegments.floorEntry(offsetToStart).getValue();
        long currentTotalSizeOfEntries = 0;
        for (JournalEntry entry : entries) {
          if (entry.getOffset().compareTo(currentSegment.getEndOffset()) > 0) {
            Offset nextSegmentStartOffset = indexSegments.higherKey(currentSegment.getStartOffset());
            currentSegment = indexSegments.get(nextSegmentStartOffset);
          }
          if (endTimeSeconds < currentSegment.getLastModifiedTimeSecs()) {
            break;
          }

          IndexValue value =
              findKey(entry.getKey(), new FileSpan(entry.getOffset(), getCurrentEndOffset(indexSegments)),
                  EnumSet.allOf(IndexEntryType.class), indexSegments);
          if (value.isDelete()) {
            messageEntries.add(
                new MessageInfo(entry.getKey(), value.getSize(), true, value.isTtlUpdate(), value.isUndelete(),
                    value.getExpiresAtMs(), null, value.getAccountId(), value.getContainerId(),
                    value.getOperationTimeInMs(), value.getLifeVersion()));
          }
          offsetEnd = entry.getOffset();
          currentTotalSizeOfEntries += value.getSize();
          if (currentTotalSizeOfEntries >= maxTotalSizeOfEntries) {
            break;
          }
        }
        // This method is only used by hard delete. Although reset key temporarily doesn't apply to hard delete, we still feed in reset key info for future use.
        IndexSegment segmentOfToken = indexSegments.floorEntry(offsetEnd).getValue();
        newToken = new StoreFindToken(offsetEnd, sessionId, incarnationId, false, segmentOfToken.getResetKey(),
            segmentOfToken.getResetKeyType(), segmentOfToken.getResetKeyLifeVersion());
      } else {
        // Case 3: offset based, but offset out of journal
        Map.Entry<Offset, IndexSegment> entry = indexSegments.floorEntry(offsetToStart);
        if (entry != null && entry.getKey() != indexSegments.lastKey()) {
          newToken = findEntriesFromSegmentStartOffset(entry.getKey(), null, messageEntries,
              new FindEntriesCondition(maxTotalSizeOfEntries, endTimeSeconds), indexSegments, false);
          if (newToken.getType().equals(FindTokenType.Uninitialized)) {
            newToken = storeToken;
          }
        } else {
          newToken = storeToken; //use the same offset as before.
        }
      }
    }
    // Filter out all the messages that are not "deleted", then update state for remaining deleted message.
    // First, filter out all the entries that are not delete
    // second, update all the remaining delete entries' state since delete might not be the final state of those entries, they can be undeleted.
    // third, filter out undelete entries after
    // fourth, eliminate duplicate deleted entries. We might have more than one versions of delete entries here.
    filterDeleteEntries(messageEntries);
    updateStateForMessages(messageEntries);
    filterDeleteEntries(messageEntries);
    eliminateDuplicates(messageEntries);
    return new FindInfo(messageEntries, newToken);
  }
Finds all the deleted entries from the given start token. The token defines the start position in the index from
where entries needs to be fetched. All the entries returned from this method have delete as it's final state, which
means if we see a Delete entry but this key is undelete later, it will not be returned.
@param token The token that signifies the start position in the index from where deleted entries need to be
             retrieved
@param maxTotalSizeOfEntries The maximum total size of entries that need to be returned. The api will try to
                             return a list of entries whose total size is close to this value.
@param endTimeSeconds The (approximate) time of the latest entry to be fetched. This is used at segment granularity.
@return The FindInfo state that contains both the list of entries and the new findtoken to start the next iteration
ambry-store/src/main/java/com/github/ambry/store/PersistentIndex.java