private void addCloudReplica(String partitionName) throws Exception {
    // Adding cloud replica occurs when replica becomes leader from standby. Hence, if this a new added replica, it
    // should be present in storage manager already.
    ReplicaId localReplica = storeManager.getReplica(partitionName);
    if (localReplica == null) {
      logger.warn("Got partition leader notification for partition {} that is not present on the node", partitionName);
      return;
    }
    PartitionId partitionId = localReplica.getPartitionId();
    Store store = storeManager.getStore(partitionId);
    if (store == null) {
      logger.warn("Unable to add cloud replica for partition {} as store for the partition is not present or started.",
          partitionName);
      return;
    }
    DataNodeId cloudDataNode = new CloudServiceDataNode(clusterMapConfig);
    CloudReplica peerCloudReplica = new CloudReplica(partitionId, cloudDataNode);
    FindTokenFactory findTokenFactory =
        tokenHelper.getFindTokenFactoryFromReplicaType(peerCloudReplica.getReplicaType());
    RemoteReplicaInfo remoteReplicaInfo =
        new RemoteReplicaInfo(peerCloudReplica, localReplica, store, findTokenFactory.getNewFindToken(),
            storeConfig.storeDataFlushIntervalSeconds * SystemTime.MsPerSec * Replication_Delay_Multiplier,
            SystemTime.getInstance(), peerCloudReplica.getDataNodeId().getPortToConnectTo());
    replicationMetrics.addMetricsForRemoteReplicaInfo(remoteReplicaInfo, trackPerDatacenterLagInMetric,
        trackPerReplicaReplicationBytes);

    // Note that for each replica on a Ambry server node, there is only one cloud replica that it will be replicating from.
    List<RemoteReplicaInfo> remoteReplicaInfos = Collections.singletonList(remoteReplicaInfo);
    PartitionInfo partitionInfo = new PartitionInfo(remoteReplicaInfos, partitionId, store, localReplica);
    partitionToPartitionInfo.put(partitionId, partitionInfo);
    mountPathToPartitionInfos.computeIfAbsent(localReplica.getMountPath(), key -> ConcurrentHashMap.newKeySet())
        .add(partitionInfo);
    logger.info("Cloud Partition {} added to {}. CloudNode {} port {}", partitionName, dataNodeId, cloudDataNode,
        cloudDataNode.getPortToConnectTo());

    // Reload replication token if exist.
    reloadReplicationTokenIfExists(localReplica, remoteReplicaInfos);

    // Add remoteReplicaInfos to {@link ReplicaThread}.
    addRemoteReplicaInfoToReplicaThread(remoteReplicaInfos, true);
    if (replicationConfig.replicationTrackPerPartitionLagFromRemote) {
      replicationMetrics.addLagMetricForPartition(partitionId, true);
    }
    logger.info("Added cloud replica for partition-{} for recovery", partitionName);
  }
Add a replica of given partition and its {@link RemoteReplicaInfo}s to backup list.
@param partitionName name of the partition of the replica to add.
@throws ReplicationException if replicas initialization failed.
ambry-cloud/src/main/java/com/github/ambry/cloud/RecoveryManager.java