List<ExchangeMetadataResponse> handleReplicaMetadataResponse(ReplicaMetadataResponse response,
      List<RemoteReplicaInfo> replicasToReplicatePerNode, DataNodeId remoteNode, List<StoreKey> storeKeysConversionLog) throws Exception {
    long startTimeInMs = time.milliseconds();
    List<ExchangeMetadataResponse> exchangeMetadataResponseList = new ArrayList<>();
    Map<StoreKey, StoreKey> remoteKeyToLocalKeyMap = batchConvertReplicaMetadataResponseKeys(response, storeKeysConversionLog);

    for (int i = 0; i < response.getReplicaMetadataResponseInfoList().size(); i++) {
      RemoteReplicaInfo remoteReplicaInfo = replicasToReplicatePerNode.get(i);
      ReplicaMetadataResponseInfo replicaMetadataResponseInfo = response.getReplicaMetadataResponseInfoList().get(i);
      responseHandler.onEvent(remoteReplicaInfo.getReplicaId(), replicaMetadataResponseInfo.getError());
      if (replicaMetadataResponseInfo.getError() == ServerErrorCode.NoError) {
        // Skip stores that were stopped during call to getReplicaMetadataResponse
        if (!remoteReplicaInfo.getLocalStore().isStarted()) {
          exchangeMetadataResponseList.add(new ExchangeMetadataResponse(ServerErrorCode.TemporarilyDisabled));
        } else {
          try {
            logger.trace("Remote node: {} Thread name: {} Remote replica: {} Token from remote: {} Replica lag: {} ",
                remoteNode, threadName, remoteReplicaInfo.getReplicaId(), replicaMetadataResponseInfo.getFindToken(),
                replicaMetadataResponseInfo.getRemoteReplicaLagInBytes());
            Set<MessageInfo> remoteMissingStoreMessages =
                getMissingStoreMessages(replicaMetadataResponseInfo, remoteNode, remoteReplicaInfo);
            processReplicaMetadataResponse(remoteMissingStoreMessages, replicaMetadataResponseInfo, remoteReplicaInfo,
                remoteNode, remoteKeyToLocalKeyMap);

            // Get the converted keys for the missing keys of this replica (to store them along with missing keys in
            // the exchange metadata response). For leader based replication, these are used during processing
            // of missing keys for non-leader replica pairs which will come later via leader<->leader replication.
            Map<StoreKey, StoreKey> remoteKeyToLocalKeySubMap = new HashMap<>();
            remoteMissingStoreMessages.forEach(remoteMissingStoreMessage -> {
              StoreKey remoteKey = remoteMissingStoreMessage.getStoreKey();
              remoteKeyToLocalKeySubMap.put(remoteKey, remoteKeyToLocalKeyMap.get(remoteKey));
            });

            ExchangeMetadataResponse exchangeMetadataResponse =
                new ExchangeMetadataResponse(remoteMissingStoreMessages, replicaMetadataResponseInfo.getFindToken(),
                    replicaMetadataResponseInfo.getRemoteReplicaLagInBytes(), remoteKeyToLocalKeySubMap, time);

            // update replication lag in ReplicaSyncUpManager
            if (replicaSyncUpManager != null
                && remoteReplicaInfo.getLocalStore().getCurrentState() == ReplicaState.BOOTSTRAP) {
              ReplicaId localReplica = remoteReplicaInfo.getLocalReplicaId();
              ReplicaId remoteReplica = remoteReplicaInfo.getReplicaId();
              boolean isSyncCompleted =
                  replicaSyncUpManager.updateReplicaLagAndCheckSyncStatus(localReplica, remoteReplica,
                      exchangeMetadataResponse.localLagFromRemoteInBytes, ReplicaState.STANDBY);
              // if catchup is completed by this update call, we can complete bootstrap in local store
              if (isSyncCompleted) {
                // complete BOOTSTRAP -> STANDBY transition
                remoteReplicaInfo.getLocalStore().setCurrentState(ReplicaState.STANDBY);
                remoteReplicaInfo.getLocalStore().completeBootstrap();
              }
            }

            // If remote token has not moved forward, wait for back off time before resending next metadata request
            if (remoteReplicaInfo.getToken().equals(exchangeMetadataResponse.remoteToken)) {
              remoteReplicaInfo.setReEnableReplicationTime(
                  time.milliseconds() + replicationConfig.replicationSyncedReplicaBackoffDurationMs);
              syncedBackOffCount.inc();
            }

            // trace replication status to track progress of recovery from cloud
            logReplicationStatus(remoteReplicaInfo, exchangeMetadataResponse);

            // There are no missing keys. We just advance the token
            if (exchangeMetadataResponse.missingStoreMessages.size() == 0) {
              advanceToken(remoteReplicaInfo, exchangeMetadataResponse);
            }

            replicationMetrics.updateLagMetricForRemoteReplica(remoteReplicaInfo,
                exchangeMetadataResponse.localLagFromRemoteInBytes);

            // Add exchangeMetadataResponse to list at the end after operations such as replicaSyncUpManager(if not null)
            // has completed update, etc. The reason is we may get exceptions in between (for ex: replicaSyncUpManager may
            // throw exception) and end up adding one more exchangeMetadataResponse associated with same RemoteReplicaInfo.
            exchangeMetadataResponseList.add(exchangeMetadataResponse);
          } catch (Exception e) {
            if (e instanceof StoreException && ((StoreException) e).getErrorCode() == StoreErrorCodes.StoreNotStarted) {
              // Must have just been stopped, just skip it and move on.
              logger.info("Local store not started for remote replica: {}", remoteReplicaInfo.getReplicaId());
              exchangeMetadataResponseList.add(new ExchangeMetadataResponse(ServerErrorCode.TemporarilyDisabled));
            } else {
              logger.error("Remote node: {} Thread name: {} Remote replica: {}", remoteNode, threadName,
                  remoteReplicaInfo.getReplicaId(), e);
              replicationMetrics.updateLocalStoreError(remoteReplicaInfo.getReplicaId());
              responseHandler.onEvent(remoteReplicaInfo.getReplicaId(), e);
              exchangeMetadataResponseList.add(new ExchangeMetadataResponse(ServerErrorCode.UnknownError));
            }
          }
        }
      } else {
        replicationMetrics.incrementReplicationErrorCount(replicatingFromRemoteColo, datacenterName);
        replicationMetrics.updateMetadataRequestError(remoteReplicaInfo.getReplicaId());
        logger.error("Remote node: {} Thread name: {} Remote replica: {} Server metadata-response error: {}",
            remoteNode, threadName, remoteReplicaInfo.getReplicaId(), replicaMetadataResponseInfo.getError());
        exchangeMetadataResponseList.add(new ExchangeMetadataResponse(replicaMetadataResponseInfo.getError()));
      }

      if (replicatingFromRemoteColo && leaderBasedReplicationAdmin != null) {
        ExchangeMetadataResponse exchangeMetadataResponse = exchangeMetadataResponseList.get(i);
        if (exchangeMetadataResponse.serverErrorCode.equals(ServerErrorCode.NoError)) {

          // If leader-based replication is enabled, store the metadata exchange received for the remote replica as
          // standby replicas will not send GET request for the missing store keys and track them from leader <->
          // leader exchanges and intra-dc replication.
          remoteReplicaInfo.setExchangeMetadataResponse(new ExchangeMetadataResponse(exchangeMetadataResponse));

          // It is possible that some missing keys found in exchange metadata response are written in parallel
          // by other replica threads since the time we calculated it. Go through the local store once more and
          // update missing keys set stored in the exchangeMetadataResponse for the remote replica.
          refreshMissingStoreMessagesForStandbyReplica(remoteReplicaInfo);
        }
      }
    }
    long processMetadataResponseTimeInMs = time.milliseconds() - startTimeInMs;
    logger.trace("Remote node: {} Thread name: {} processMetadataResponseTime: {}", remoteNode, threadName,
        processMetadataResponseTimeInMs);
    return exchangeMetadataResponseList;
  }
Handle {@link ReplicaMetadataResponse} returned from the given remote {@link DataNodeId}. It will find all the missing
keys in the local blobstore that exist in the remote data node. And for those keys that exist in both blob stores, it
will apply update to local blob stores if needed.
@param response The {@link ReplicaMetadataResponse}.
@param replicasToReplicatePerNode
@param remoteNode The remote {@link DataNodeId}.
@param storeKeysConversionLog list to store storeKeys that were converted
@return A list of {@link ExchangeMetadataResponse}.
@throws IOException
ambry-replication/src/main/java/com/github/ambry/replication/ReplicaThread.java