public void retrieveReplicaTokensAndPersistIfNecessary(String mountPath) throws ReplicationException, IOException {
    boolean tokenWasReset = false;
    long readStartTimeMs = time.milliseconds();
    List<RemoteReplicaInfo.ReplicaTokenInfo> tokenInfoList = persistor.retrieve(mountPath);

    for (RemoteReplicaInfo.ReplicaTokenInfo tokenInfo : tokenInfoList) {
      String hostname = tokenInfo.getHostname();
      int port = tokenInfo.getPort();
      PartitionId partitionId = tokenInfo.getPartitionId();
      FindToken token = tokenInfo.getReplicaToken();
      // update token
      PartitionInfo partitionInfo = partitionToPartitionInfo.get(tokenInfo.getPartitionId());
      if (partitionInfo != null) {
        boolean updatedToken = false;
        for (RemoteReplicaInfo remoteReplicaInfo : partitionInfo.getRemoteReplicaInfos()) {
          if (isTokenForRemoteReplicaInfo(remoteReplicaInfo, tokenInfo)) {
            logger.info("Read token for partition {} remote host {} port {} token {}", partitionId, hostname, port,
                token);
            if (partitionInfo.getStore().isEmpty() || partitionInfo.getStore().hasPartialRecovery()) {
              // If the local replica is empty, it could have been newly created. In this case, the offset in
              // every peer replica which the local replica lags from should be set to 0, so that the local
              // replica starts fetching from the beginning of the peer. The totalBytes the peer read from the
              // local replica should also be set to 0. During initialization these values are already set to 0,
              // so we let them be.
              //
              // If the local replica has partial recovery, the local replica should start fetching from the beginning
              // of the peer to just be safe.
              tokenWasReset = true;
              logTokenReset(partitionId, hostname, port, token);
            } else {
              remoteReplicaInfo.initializeTokens(token);
              remoteReplicaInfo.setTotalBytesReadFromLocalStore(tokenInfo.getTotalBytesReadFromLocalStore());
            }
            updatedToken = true;
            break;
          }
        }
        if (!updatedToken) {
          logger.warn("Persisted remote replica host {} and port {} not present in new cluster ", hostname, port);
        }
      } else {
        // If this partition was not found in partitionToPartitionInfo, it means that the local store corresponding
        // to this partition could not be started. In such a case, the tokens for its remote replicas should be
        // reset.
        tokenWasReset = true;
        logTokenReset(partitionId, hostname, port, token);
      }
    }
    replicationMetrics.remoteReplicaTokensRestoreTime.update(time.milliseconds() - readStartTimeMs);

    if (tokenWasReset) {
      // We must ensure that the token file is persisted if any of the tokens in the file got reset. We need to do
      // this before an associated store takes any writes, to avoid the case where a store takes writes and persists it,
      // before the replica token file is persisted after the reset.
      if (persistor != null) {
        persistor.write(mountPath, false);
      } else {
        logger.warn("Unable to persist after token reset, persistor is null");
      }
    }
  }
Reads the replica tokens from storage, populates the Remote replica info,
and re-persists the tokens if they have been reset.
@param mountPath The mount path where the replica tokens are stored
@throws ReplicationException
@throws IOException
ambry-replication/src/main/java/com/github/ambry/replication/ReplicationEngine.java