@Test
  public void compactDeleteTombstoneTwiceTest() throws Exception {
    assumeTrue(purgeDeleteTombstone);
    refreshState(false, true, false);
    List<LogSegmentName> segmentsUnderCompaction = getLogSegments(0, 2);
    CompactionDetails details = new CompactionDetails(state.time.milliseconds(), segmentsUnderCompaction, null);
    List<MockReplicaId> localAndPeerReplicas = generateLocalAndPeerReplicas();
    ScheduledExecutorService scheduler = Utils.newScheduler(1, false);
    MockIdFactory keyFactory = new MockIdFactory();
    RemoteTokenTracker tokenTracker =
        new RemoteTokenTracker(localAndPeerReplicas.get(0), scheduler, keyFactory, state.time);
    MockId tombstone1 = state.permanentDeleteTombstones.get(0);
    MockId tombstone2 = state.permanentDeleteTombstones.get(1);
    IndexValue deleteIndexValue2 = state.index.findKey(tombstone2);
    IndexSegment indexSegment2 = state.index.getIndexSegments().floorEntry(deleteIndexValue2.getOffset()).getValue();
    MockId keyInToken = (MockId) indexSegment2.iterator().next().getKey();

    UUID invalidIncarnationId;
    do {
      invalidIncarnationId = UUID.randomUUID();
    } while (invalidIncarnationId.equals(state.incarnationId));
    // invalid token with other incarnation id
    StoreFindToken invalidToken =
        new StoreFindToken(keyInToken, indexSegment2.getStartOffset(), state.sessionId, invalidIncarnationId,
            indexSegment2.getResetKey(), indexSegment2.getResetKeyType(), indexSegment2.getResetKeyLifeVersion());
    StoreFindToken peerToken2 =
        new StoreFindToken(keyInToken, indexSegment2.getStartOffset(), state.sessionId, state.incarnationId,
            indexSegment2.getResetKey(), indexSegment2.getResetKeyType(), indexSegment2.getResetKeyLifeVersion());
    MockReplicaId peerReplica1 = localAndPeerReplicas.get(1);
    MockReplicaId peerReplica2 = localAndPeerReplicas.get(2);
    tokenTracker.updateTokenFromPeerReplica(invalidToken, peerReplica1.getDataNodeId().getHostname(),
        peerReplica1.getReplicaPath());
    tokenTracker.updateTokenFromPeerReplica(peerToken2, peerReplica2.getDataNodeId().getHostname(),
        peerReplica2.getReplicaPath());
    // initiate compaction
    compactor = getCompactor(state.log, DISK_IO_SCHEDULER, tokenTracker, false);
    compactor.initialize(state.index);
    try {
      compactor.compact(details, bundleReadBuffer);
    } finally {
      compactor.close(0);
    }
    // both tombstones should exist
    assertNotNull("Delete tombstone should be present", state.index.findKey(tombstone1));
    assertNotNull("Delete tombstone should be present", state.index.findKey(tombstone2));

    // update remote token tracker with journal based tokens and compact again.
    IndexSegment lastIndexSegment = state.index.getIndexSegments().lastEntry().getValue();
    peerToken2 = new StoreFindToken(lastIndexSegment.getStartOffset(), state.sessionId, state.incarnationId, true,
        lastIndexSegment.getResetKey(), lastIndexSegment.getResetKeyType(), lastIndexSegment.getResetKeyLifeVersion());
    tokenTracker.updateTokenFromPeerReplica(peerToken2, peerReplica1.getDataNodeId().getHostname(),
        peerReplica1.getReplicaPath());
    tokenTracker.updateTokenFromPeerReplica(peerToken2, peerReplica2.getDataNodeId().getHostname(),
        peerReplica2.getReplicaPath());
    // initiate compaction
    segmentsUnderCompaction = getLogSegments(0, 2);
    details = new CompactionDetails(state.time.milliseconds(), segmentsUnderCompaction, null);
    compactor = getCompactor(state.log, DISK_IO_SCHEDULER, tokenTracker, false);
    compactor.initialize(state.index);
    try {
      compactor.compact(details, bundleReadBuffer);
    } finally {
      compactor.close(0);
    }
    tokenTracker.close();

    // both tombstones should be compacted
    assertNull("Delete tombstone should be compacted", state.index.findKey(tombstone1));
    assertNull("Delete tombstone should be compacted", state.index.findKey(tombstone2));
  }
Tests compacting delete tombstone with both invalid and journal based tokens.
@throws Exception
ambry-store/src/test/java/com/github/ambry/store/BlobStoreCompactorTest.java