@Test
  public void deleteTombstoneIgnoreOfflinePeers() 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);
    tokenTracker.close(); // stop the background persist. will explicitly call persistToken later.
    // Generate token that is past position of 1st delete tombstone and haven't reached position of 2nd tombstone.
    MockId tombstone1 = state.permanentDeleteTombstones.get(0);
    MockId tombstone2 = state.permanentDeleteTombstones.get(1);
    IndexValue deleteIndexValue1 = state.index.findKey(tombstone1);
    IndexSegment indexSegment1 = state.index.getIndexSegments().floorEntry(deleteIndexValue1.getOffset()).getValue();
    IndexSegment segmentBehindSegment1 =
        state.index.getIndexSegments().higherEntry(indexSegment1.getStartOffset()).getValue();
    MockId keyInToken = (MockId) segmentBehindSegment1.iterator().next().getKey();
    // We have two tokens now, first token is an uninitialized token, second token is the one point to somewhere between
    // first tombstone and second tombstone.
    StoreFindToken peerToken1 = new StoreFindToken();
    StoreFindToken peerToken2 =
        new StoreFindToken(keyInToken, segmentBehindSegment1.getStartOffset(), state.sessionId, state.incarnationId,
            segmentBehindSegment1.getResetKey(), segmentBehindSegment1.getResetKeyType(),
            segmentBehindSegment1.getResetKeyLifeVersion());
    // update token associated with peer replica
    MockReplicaId peerReplica1 = localAndPeerReplicas.get(1);
    MockReplicaId peerReplica2 = localAndPeerReplicas.get(2);
    tokenTracker.updateTokenFromPeerReplica(peerToken1, peerReplica1.getDataNodeId().getHostname(),
        peerReplica1.getReplicaPath());
    tokenTracker.updateTokenFromPeerReplica(peerToken2, peerReplica2.getDataNodeId().getHostname(),
        peerReplica2.getReplicaPath());
    tokenTracker.persistToken();
    compactor = getCompactor(state.log, DISK_IO_SCHEDULER, tokenTracker, false);
    compactor.initialize(state.index);
    try {
      compactor.compact(details, bundleReadBuffer);
    } finally {
      compactor.close(0);
    }

    // both delete tombstones should be not compacted
    assertNotNull(state.index.findKey(tombstone1));
    assertNotNull(state.index.findKey(tombstone2));

    // Now move the timestamp forward for 7+1 days.
    final int offlineReplicasDays = 7;
    state.time.sleep(TimeUnit.DAYS.toMillis(offlineReplicasDays + 1));
    // Update the timestamp for second peer
    deleteIndexValue1 = state.index.findKey(tombstone1);
    indexSegment1 = state.index.getIndexSegments().floorEntry(deleteIndexValue1.getOffset()).getValue();
    segmentBehindSegment1 = state.index.getIndexSegments().higherEntry(indexSegment1.getStartOffset()).getValue();
    keyInToken = (MockId) segmentBehindSegment1.iterator().next().getKey();
    peerToken2 =
        new StoreFindToken(keyInToken, segmentBehindSegment1.getStartOffset(), state.sessionId, state.incarnationId,
            segmentBehindSegment1.getResetKey(), segmentBehindSegment1.getResetKeyType(),
            segmentBehindSegment1.getResetKeyLifeVersion());
    tokenTracker.updateTokenFromPeerReplica(peerToken2, peerReplica2.getDataNodeId().getHostname(),
        peerReplica2.getReplicaPath());
    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);
    }
    // both delete tombstones should be not compacted since the configuration to ignore offline replicas is not turned on
    assertNotNull(state.index.findKey(tombstone1));
    assertNotNull(state.index.findKey(tombstone2));

    // Now turn on the configuration
    deleteIndexValue1 = state.index.findKey(tombstone1);
    indexSegment1 = state.index.getIndexSegments().floorEntry(deleteIndexValue1.getOffset()).getValue();
    segmentBehindSegment1 = state.index.getIndexSegments().higherEntry(indexSegment1.getStartOffset()).getValue();
    keyInToken = (MockId) segmentBehindSegment1.iterator().next().getKey();
    peerToken2 =
        new StoreFindToken(keyInToken, segmentBehindSegment1.getStartOffset(), state.sessionId, state.incarnationId,
            segmentBehindSegment1.getResetKey(), segmentBehindSegment1.getResetKeyType(),
            segmentBehindSegment1.getResetKeyLifeVersion());
    tokenTracker.updateTokenFromPeerReplica(peerToken2, peerReplica2.getDataNodeId().getHostname(),
        peerReplica2.getReplicaPath());
    state.properties.setProperty(StoreConfig.storeCompactionIgnorePeersUnavailableForDaysName,
        String.valueOf(offlineReplicasDays));
    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);
    }
    // first replica should be compacted
    assertNull("First delete tombstone " + tombstone1 + " should be compacted", state.index.findKey(tombstone1));
    assertNotNull(state.index.findKey(tombstone2));

    // Sleep for another 7 days so peer2 would become offline as well
    state.time.sleep(TimeUnit.DAYS.toMillis(offlineReplicasDays + 1));
    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);
    }
    // first replica should be compacted
    assertNull("First delete tombstone " + tombstone1 + " should be compacted", state.index.findKey(tombstone1));
    assertNull(state.index.findKey(tombstone2));
  }
Test the case where peers went offline for a long time and they should be ignored when checking if a delete tombstone
is still valid
@throws Exception
ambry-store/src/test/java/com/github/ambry/store/BlobStoreCompactorTest.java