List<IndexEntry> getValidIndexEntriesForIndexSegment(Offset indexSegmentStartOffset, long deleteReferenceTimeMs,
      long expiryReferenceTimeMs, FileSpan fileSpanUnderCompaction, DeleteTombstoneStats.Builder builder,
      Pair<Set<MockId>, Set<MockId>> expiredDeletes, boolean invalidateExpiredDelete,
      BlobStoreStats.IndexEntryAction action) {
    List<IndexEntry> validEntries = new ArrayList<>();
    if (referenceIndex.containsKey(indexSegmentStartOffset)) {
      for (Map.Entry<MockId, TreeSet<IndexValue>> indexSegmentEntry : referenceIndex.get(indexSegmentStartOffset)
          .entrySet()) {
        MockId key = indexSegmentEntry.getKey();
        NavigableSet<IndexValue> values = indexSegmentEntry.getValue();
        IndexValue latestValue = getExpectedValue(key, false);
        Iterator<IndexValue> iter = values.iterator();
        while (iter.hasNext()) {
          IndexValue currentValue = iter.next();
          IndexEntry currentEntry = new IndexEntry(key, currentValue);
          boolean isValid = false;
          if (currentValue.isUndelete()) {
            if (latestValue.isUndelete() && latestValue.getLifeVersion() == currentValue.getLifeVersion()) {
              validEntries.add(currentEntry);
              isValid = true;
            }
          } else if (currentValue.isDelete()) {
            if (latestValue.isDelete() && latestValue.getLifeVersion() == currentValue.getLifeVersion()) {
              // check if this is a delete tombstone left by compaction
              isValid = true;
              try {
                if (getExpectedValue(key, true) == null) {
                  if (currentValue.getExpiresAtMs() == Utils.Infinite_Time || currentValue.isTtlUpdate()) {
                    // TODO check validity of permanent delete tombstone
                    builder.permanentCountInc()
                        .permanentSizeInc(currentValue.getSize(), latestValue.getOperationTimeInMs());
                  } else {
                    builder.expiredCountInc().expiredSizeInc(currentValue.getSize());
                    if (invalidateExpiredDelete && currentValue.getExpiresAtMs() < expiryReferenceTimeMs) {
                      isValid = false;
                      expiredDeletes.getFirst().add(key);
                    }
                  }
                } else {
                  // if this is a delete with PUT, we check if it has expired and track it in a separate set
                  if (invalidateExpiredDelete && currentValue.getExpiresAtMs() != Utils.Infinite_Time
                      && currentValue.getExpiresAtMs() < expiryReferenceTimeMs) {
                    expiredDeletes.getSecond().add(key);
                  }
                }
              } catch (Exception e) {
                // do nothing for now
              }
              if (isValid) {
                validEntries.add(currentEntry);
              }
            }
          } else if (currentValue.isTtlUpdate()) {
            IndexValue putValue = getExpectedValue(key, EnumSet.of(PersistentIndex.IndexEntryType.PUT), null);
            if (putValue == null) {
              continue;
            }
            if (latestValue.isDelete()) {
              if (latestValue.getOperationTimeInMs() >= deleteReferenceTimeMs
                  || isTtlUpdateEntryValidWhenFinalStateIsDeleteAndRetention(key, currentValue,
                  fileSpanUnderCompaction)) {
                validEntries.add(currentEntry);
                isValid = true;
              }
            } else {
              validEntries.add(currentEntry);
              isValid = true;
            }
          } else {
            // current index value is PUT
            if (!isExpiredAt(key, expiryReferenceTimeMs)) {
              if (latestValue.isPut()) {
                validEntries.add(currentEntry);
                isValid = true;
              } else if (latestValue.isDelete()) {
                if (latestValue.getOperationTimeInMs() >= deleteReferenceTimeMs) {
                  validEntries.add(currentEntry);
                  isValid = true;
                }
              } else {
                validEntries.add(currentEntry);
                isValid = true;
              }
            }
          }
          if (action != null) {
            action.accept(currentEntry, isValid);
          }
        }
      }
    }
    return validEntries;
  }
Gets all the valid index entries (taking into account different reference times) in the index segment with start
offset {@code indexSegmentStartOffset}.
@param indexSegmentStartOffset the start offset of the {@link IndexSegment} from which valid index entries are
                               required.
@param deleteReferenceTimeMs the reference time in ms until which deletes are relevant.
@param expiryReferenceTimeMs the reference time in ms until which expirations are relevant
@param fileSpanUnderCompaction only useful for compaction tests to help determine whether update records will
                               cleaned up. All other use cases should use {@code null}.
@param builder a builder that tracks stats related delete tombstones in log segments.
@param expiredDeletes a pair of two sets of deletes. The first one contains expired delete tombstones (with no
                      associated PUT); the second one includes expired deletes currently with PUTs ahead of them.
@param invalidateExpiredDelete whether to invalidate expired delete.
@return all the valid index entries valid in the index segment with start offset {@code indexSegmentStartOffset}.
ambry-store/src/test/java/com/github/ambry/store/CuratedLogIndexState.java