@Override
  public void advanceToken(RemoteReplicaInfo remoteReplicaInfo, ExchangeMetadataResponse exchangeMetadataResponse) {
    StoreFindToken oldToken = (StoreFindToken) remoteReplicaInfo.getToken();
    // The parent method sets in-memory token
    super.advanceToken(remoteReplicaInfo, exchangeMetadataResponse);
    // The lag can be -1 at times, so just round up to 0 and move on
    azureStorageContainerMetricsCollector.setPartitionReplicaLag(
        remoteReplicaInfo,
        Math.max(0, exchangeMetadataResponse.getLocalLagFromRemoteInBytes()));
    StoreFindToken token = (StoreFindToken) remoteReplicaInfo.getToken();
    if (token == null) {
      azureMetrics.replicaTokenWriteErrorCount.inc();
      logger.error("Null token for replica {}", remoteReplicaInfo);
      return;
    }
    if (token.equals(oldToken)) {
      logger.trace("Skipping token upload due to unchanged token, oldToken = {}, newToken = {}", oldToken, token);
      return;
    }
    logger.trace("Uploading token = {} for replica = {}", token, remoteReplicaInfo);
    // Table entity = Table row
    // ======================================================================================
    // | partition-key | row-key | tokenType | logSegment | offset | storeKey | binaryToken |
    // ======================================================================================
    // | partition     | host1   | Journal   | 3_14       | 12     | none     | AAASLKJDFX  |
    // | partition     | host2   | Index     | 2_12       | 32     | AAEWsXZ  | AAAEWODSDS  |
    // ======================================================================================
    String partitionKey = String.valueOf(remoteReplicaInfo.getReplicaId().getPartitionId().getId());
    String rowKey = remoteReplicaInfo.getReplicaId().getDataNodeId().getHostname();
    TableEntity entity = new TableEntity(partitionKey, rowKey)
        .addProperty(VcrReplicationManager.BACKUP_NODE, dataNodeId.getHostname())
        .addProperty(VcrReplicationManager.TOKEN_TYPE,
            token.getType().toString())
        .addProperty(VcrReplicationManager.LOG_SEGMENT,
            token.getOffset() == null ? "none" : token.getOffset().getName().toString())
        .addProperty(VcrReplicationManager.OFFSET,
            token.getOffset() == null ? "none" : token.getOffset().getOffset())
        .addProperty(VcrReplicationManager.STORE_KEY,
            token.getStoreKey() == null ? "none" : token.getStoreKey().getID())
        .addProperty(VcrReplicationManager.BINARY_TOKEN,
            token.toBytes());
    // Now persist the token in cloud
    if (cloudDestination.upsertTableEntity(azureTableNameReplicaTokens, entity)) {
      azureMetrics.replicaTokenWriteRate.mark();
    } else {
      azureMetrics.replicaTokenWriteErrorCount.inc();
    }
  }
Persists token to cloud in each replication cycle
@param remoteReplicaInfo Remote replica info object
@param exchangeMetadataResponse Metadata object exchanged between replicas
ambry-cloud/src/main/java/com/github/ambry/cloud/VcrReplicaThread.java