@Test
  public void remoteTokenPersistTest() throws Exception {
    assumeTrue(purgeDeleteTombstone);
    refreshState(false, true, false);
    List<MockReplicaId> localAndPeerReplicas = generateLocalAndPeerReplicas();
    MockIdFactory keyFactory = new MockIdFactory();
    ScheduledExecutorService scheduler = Utils.newScheduler(1, true);

    // Prepare for the token to update
    // Generate tokens for peer replicas and make sure they are both 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);
    IndexValue deleteIndexValue2 = state.index.findKey(tombstone2);
    IndexSegment indexSegment1 = state.index.getIndexSegments().floorEntry(deleteIndexValue1.getOffset()).getValue();
    IndexSegment indexSegment2 = state.index.getIndexSegments().floorEntry(deleteIndexValue2.getOffset()).getValue();
    MockId keyInToken1 = (MockId) indexSegment1.iterator().next().getKey();
    MockId keyInToken2 = (MockId) indexSegment2.iterator().next().getKey();
    StoreFindToken peerToken1 =
        new StoreFindToken(keyInToken1, indexSegment1.getStartOffset(), state.sessionId, state.incarnationId,
            indexSegment1.getResetKey(), indexSegment1.getResetKeyType(), indexSegment1.getResetKeyLifeVersion());
    StoreFindToken peerToken2 =
        new StoreFindToken(keyInToken2, indexSegment2.getStartOffset(), state.sessionId, state.incarnationId,
            indexSegment2.getResetKey(), indexSegment2.getResetKeyType(), indexSegment2.getResetKeyLifeVersion());
    // update token associated with peer replica
    MockReplicaId peerReplica1 = localAndPeerReplicas.get(1);
    MockReplicaId peerReplica2 = localAndPeerReplicas.get(2);

    // Case 1: Test token persist file doesn't exist
    RemoteTokenTracker tokenTracker =
        new RemoteTokenTracker(localAndPeerReplicas.get(0), scheduler, keyFactory, state.time);
    Map<String, Pair<Long, FindToken>> notPersistedToken = tokenTracker.getPeerReplicaAndToken();
    assertEquals(2, notPersistedToken.size());
    for (Map.Entry<String, Pair<Long, FindToken>> entry : notPersistedToken.entrySet()) {
      assertEquals(FindTokenType.Uninitialized, entry.getValue().getSecond().getType());
    }
    assertEquals(true, tokenTracker.persistToken());    // persist the token

    // Case 2: Although the file exist, no meaningful token content. The token timestamp should be the same as before
    tokenTracker = new RemoteTokenTracker(localAndPeerReplicas.get(0), scheduler, keyFactory, state.time);
    Map<String, Pair<Long, FindToken>> persistedEmptyToken = tokenTracker.getPeerReplicaAndToken();
    assertEquals(2, persistedEmptyToken.size());
    for (Map.Entry<String, Pair<Long, FindToken>> entry : persistedEmptyToken.entrySet()) {
      Pair<Long, FindToken> before = notPersistedToken.get(entry.getKey());
      assertEquals(entry.getValue().getFirst(), before.getFirst());
      assertEquals(FindTokenType.Uninitialized, entry.getValue().getSecond().getType());
    }
    assertEquals(notPersistedToken, persistedEmptyToken);
    // advance time
    state.time.sleep(10);
    // update peerReplica1's token
    tokenTracker.updateTokenFromPeerReplica(peerToken1, peerReplica1.getDataNodeId().getHostname(),
        peerReplica1.getReplicaPath());
    assertEquals(true, tokenTracker.persistToken());

    // Case 3: peerReplica1 should have the valid token
    tokenTracker = new RemoteTokenTracker(localAndPeerReplicas.get(0), scheduler, keyFactory, state.time);
    Map<String, Pair<Long, FindToken>> oneValidToken = tokenTracker.getPeerReplicaAndToken();
    assertEquals(2, oneValidToken.size());
    for (Map.Entry<String, Pair<Long, FindToken>> entry : oneValidToken.entrySet()) {
      Pair<Long, FindToken> before = persistedEmptyToken.get(entry.getKey());
      if (entry.getKey().startsWith(peerReplica1.getDataNodeId().getHostname())) {
        // token is updated
        assertTrue(entry.getValue().getFirst() > before.getFirst());
        assertEquals(entry.getValue().getSecond(), peerToken1);
      } else {
        assertEquals(entry.getValue().getFirst(), before.getFirst());
        assertEquals(FindTokenType.Uninitialized, entry.getValue().getSecond().getType());
      }
    }
    state.time.sleep(10);
    // update peerReplica2's token as well
    tokenTracker.updateTokenFromPeerReplica(peerToken2, peerReplica2.getDataNodeId().getHostname(),
        peerReplica2.getReplicaPath());
    assertEquals(true, tokenTracker.persistToken());

    // Case 4: both tokens are updated
    tokenTracker = new RemoteTokenTracker(localAndPeerReplicas.get(0), scheduler, keyFactory, state.time);
    Map<String, Pair<Long, FindToken>> twoValidTokens = tokenTracker.getPeerReplicaAndToken();
    assertEquals(2, twoValidTokens.size());
    for (Map.Entry<String, Pair<Long, FindToken>> entry : twoValidTokens.entrySet()) {
      Pair<Long, FindToken> before = oneValidToken.get(entry.getKey());
      if (entry.getKey().startsWith(peerReplica2.getDataNodeId().getHostname())) {
        // token is updated
        assertTrue(entry.getValue().getFirst() > before.getFirst());
        assertEquals(entry.getValue().getSecond(), peerToken2);
      } else {
        assertEquals(entry.getValue().getFirst(), before.getFirst());
        assertEquals(entry.getValue().getSecond(), peerToken1);
      }
    }
    // corrupt the persist file
    String filePath = tokenTracker.getPersistFilePath();
    File file = new File(filePath);
    FileOutputStream fileStream = new FileOutputStream(file);
    DataOutputStream writer = new DataOutputStream(fileStream);
    // version part.
    writer.writeShort(765);
    fileStream.getChannel().force(true);
    writer.close();

    // Case 5: Corrupted the file
    long currentTime = state.time.milliseconds();
    tokenTracker = new RemoteTokenTracker(localAndPeerReplicas.get(0), scheduler, keyFactory, state.time);
    Map<String, Pair<Long, FindToken>> corrupted = tokenTracker.getPeerReplicaAndToken();
    assertEquals(2, corrupted.size());
    for (Map.Entry<String, Pair<Long, FindToken>> entry : corrupted.entrySet()) {
      assertEquals(FindTokenType.Uninitialized, entry.getValue().getSecond().getType());
      assertTrue(entry.getValue().getFirst() >= currentTime);
    }
    assertEquals(true, tokenTracker.persistToken());    // persist the token

    // Case 6: The file is back to normal.
    tokenTracker = new RemoteTokenTracker(localAndPeerReplicas.get(0), scheduler, keyFactory, state.time);
    Map<String, Pair<Long, FindToken>> newToken = tokenTracker.getPeerReplicaAndToken();
    assertEquals(2, newToken.size());
    for (Map.Entry<String, Pair<Long, FindToken>> entry : newToken.entrySet()) {
      Pair<Long, FindToken> before = corrupted.get(entry.getKey());
      assertEquals(FindTokenType.Uninitialized, entry.getValue().getSecond().getType());
      assertEquals(before.getFirst(), entry.getValue().getFirst());       // same time stamp
    }
    assertEquals(true, tokenTracker.persistToken());

    // Case 7: Remove one replica
    MockReplicaId modifiedLocalReplica = localAndPeerReplicas.get(0);
    modifiedLocalReplica.setPeerReplicas(Collections.singletonList(peerReplica1));
    tokenTracker = new RemoteTokenTracker(modifiedLocalReplica, scheduler, keyFactory, state.time);
    Map<String, Pair<Long, FindToken>> onePeerOnlyToken = tokenTracker.getPeerReplicaAndToken();
    assertEquals(1, onePeerOnlyToken.size()); // only have one entry
    for (Map.Entry<String, Pair<Long, FindToken>> entry : onePeerOnlyToken.entrySet()) {
      Pair<Long, FindToken> before = newToken.get(entry.getKey());
      assertEquals(FindTokenType.Uninitialized, entry.getValue().getSecond().getType());
      assertEquals(before.getFirst(), entry.getValue().getFirst()); // same time stamp
      assertTrue(entry.getKey().startsWith(peerReplica1.getDataNodeId().getHostname()));
    }
    assertEquals(true, tokenTracker.persistToken());
  }
Tests RemoteTokenTracker persist the remote tokens in different case.
@throws Exception
ambry-store/src/test/java/com/github/ambry/store/BlobStoreCompactorTest.java