@Test
  public void ttlUpdateSpecificTest() throws Exception {
    // ensure that puts are not cleaned up if they have ttl updates
    Pair<List<MockId>, Long> idsAndExpiryTimeMs = createStateWithPutAndTtlUpdate();
    assertTrue("Current time should be beyond expiry time of blobs",
        state.time.milliseconds() > idsAndExpiryTimeMs.getSecond());
    List<LogSegmentName> segmentsUnderCompaction = getLogSegments(0, state.index.getLogSegmentCount() - 1);
    compactAndVerify(segmentsUnderCompaction, state.time.milliseconds(), false);

    // compact everything
    // using expire time as the ref time
    long expiryTimeMs = createStateWithPutTtlUpdateAndDelete();
    segmentsUnderCompaction = getLogSegments(0, state.index.getLogSegmentCount() - 1);
    compactAndVerify(segmentsUnderCompaction, expiryTimeMs, true);

    // using delete time as the ref time
    createStateWithPutTtlUpdateAndDelete();
    state.advanceTime(TimeUnit.SECONDS.toMillis(1));
    segmentsUnderCompaction = getLogSegments(0, state.index.getLogSegmentCount() - 1);
    compactAndVerify(segmentsUnderCompaction, state.time.milliseconds(), true);

    // compact everything except the first log segment (all the TTL updates and deletes will be retained - no change)
    // using expire time as the ref time
    expiryTimeMs = createStateWithPutTtlUpdateAndDelete();
    segmentsUnderCompaction = getLogSegments(1, state.index.getLogSegmentCount() - 2);
    compactAndVerify(segmentsUnderCompaction, expiryTimeMs, false);

    // using delete time as the ref time
    createStateWithPutTtlUpdateAndDelete();
    state.advanceTime(TimeUnit.SECONDS.toMillis(1));
    segmentsUnderCompaction = getLogSegments(1, state.index.getLogSegmentCount() - 2);
    compactAndVerify(segmentsUnderCompaction, state.time.milliseconds(), false);

    // segment that has only ttl updates and deletes (no corresponding puts). All the ttl updates should be cleaned up
    Set<MockId> ids = createStateWithTtlUpdatesAndDeletes();
    state.advanceTime(TimeUnit.SECONDS.toMillis(1));
    segmentsUnderCompaction = getLogSegments(0, state.index.getLogSegmentCount() - 1);
    compactAndVerify(segmentsUnderCompaction, state.time.milliseconds(), true);

    // there should be no ttl updates in the final index (but the deletes should be there)
    Set<MockId> seenIds = new HashSet<>();
    List<IndexEntry> indexEntries = new ArrayList<>();
    FindEntriesCondition condition = new FindEntriesCondition(Long.MAX_VALUE);
    AtomicLong currentTotalSize = new AtomicLong(0);
    for (IndexSegment segment : state.index.getIndexSegments().values()) {
      if (segment.getLogSegmentName().getGeneration() == 0) {
        break;
      }
      segment.getIndexEntriesSince(null, condition, indexEntries, currentTotalSize, false, false);
    }
    indexEntries.forEach(entry -> {
      assertTrue("There cannot be a non-delete entry", entry.getValue().isDelete());
      assertTrue("Every key should be seen only once", seenIds.add((MockId) entry.getKey()));
    });
    assertEquals("All ids not present", ids, seenIds);
  }
Tests cases specific to TTL updates
@throws Exception
ambry-store/src/test/java/com/github/ambry/store/BlobStoreCompactorTest.java