void processReplicaMetadataResponse(Set<MessageInfo> missingRemoteStoreMessages,
      ReplicaMetadataResponseInfo replicaMetadataResponseInfo, RemoteReplicaInfo remoteReplicaInfo,
      DataNodeId remoteNode, Map<StoreKey, StoreKey> remoteKeyToLocalKeyMap) throws StoreException {
    long startTime = time.milliseconds();
    List<MessageInfo> messageInfoList = replicaMetadataResponseInfo.getMessageInfoList();
    for (MessageInfo messageInfo : messageInfoList) {
      BlobId blobId = (BlobId) messageInfo.getStoreKey();
      if (remoteReplicaInfo.getLocalReplicaId().getPartitionId().compareTo(blobId.getPartition()) != 0) {
        String errorMessage = String.format(
            "Blob id %s is not in the expected partition Actual partition %s " + " Expected partition %s"
                + "Replica %s", blobId.getID(), blobId.getPartition(),
            remoteReplicaInfo.getLocalReplicaId().getPartitionId(), remoteReplicaInfo.getReplicaId());
        replicationMetrics.unexpectedBlobIdError.inc();
        if (replicationConfig.replicationIgnoreUnexpectedBlobIdError) {
          logger.error(errorMessage);
          missingRemoteStoreMessages.remove(messageInfo);
          continue;
        } else {
          throw new IllegalStateException(errorMessage);
        }
      }
      BlobId localKey = (BlobId) remoteKeyToLocalKeyMap.get(messageInfo.getStoreKey());
      if (localKey == null) {
        missingRemoteStoreMessages.remove(messageInfo);
        logger.trace("Remote node: {} Thread name: {} Remote replica: {} Remote key deprecated locally: {}", remoteNode,
            threadName, remoteReplicaInfo.getReplicaId(), messageInfo.getStoreKey());
      } else if (missingRemoteStoreMessages.contains(messageInfo)) {
        // The key is missing in the local store, we either send get request to fetch the content of this key, or does
        // nothing if the key is deleted or expired remotely.
        if (messageInfo.isDeleted()) {
          // if the key is not present locally and if the remote replica has the message in deleted state,
          // it is not considered missing locally.
          missingRemoteStoreMessages.remove(messageInfo);
          logger.trace(
              "Remote node: {} Thread name: {} Remote replica: {} Key in deleted state remotely: {} Local key: {}",
              remoteNode, threadName, remoteReplicaInfo.getReplicaId(), messageInfo.getStoreKey(), localKey);
          // A Repair event for Delete signifies that a Delete message was received from the remote and it is fired
          // as long as the Delete is guaranteed to have taken effect locally.
          if (notification != null) {
            notification.onBlobReplicaDeleted(dataNodeId.getHostname(), dataNodeId.getPort(), localKey.getID(),
                BlobReplicaSourceType.REPAIRED);
          }
        } else if (messageInfo.isExpired()) {
          // if the key is not present locally and if the remote replica has the key as expired,
          // it is not considered missing locally.
          missingRemoteStoreMessages.remove(messageInfo);
          logger.trace("Remote node: {} Thread name: {} Remote replica: {} Key in expired state remotely {}",
              remoteNode, threadName, remoteReplicaInfo.getReplicaId(), localKey);
        }
      } else {
        // The key is present in the local store. Compare blob properties (ttl_update, delete, undelete fields) in
        // received message info with blob in local store and apply updates as needed. For ex, mark local blob for deletion if
        // it is deleted in the remote store and not deleted yet locally.

        // if the blob is from deprecated container, then nothing needs to be done.
        if (replicationConfig.replicationContainerDeletionEnabled && skipPredicate != null && skipPredicate.test(
            messageInfo)) {
          continue;
        }
        applyUpdatesToBlobInLocalStore(messageInfo, remoteReplicaInfo, localKey);
      }
    }
    if (replicatingFromRemoteColo) {
      replicationMetrics.interColoProcessMetadataResponseTime.get(datacenterName)
          .update(time.milliseconds() - startTime);
    } else {
      replicationMetrics.intraColoProcessMetadataResponseTime.update(time.milliseconds() - startTime);
    }
  }
Takes the missing store messages and the message list from the remote store and identifies ones that are deleted
on the remote store and updates them locally. Also, if the message that is missing is deleted in the remote
store, we remove the message from the list of missing messages
@param missingRemoteStoreMessages The list of messages missing from the local store
@param replicaMetadataResponseInfo The replica metadata response from the remote store
@param remoteReplicaInfo The remote replica that is being replicated from
@param remoteNode The remote node from which replication needs to happen
@param remoteKeyToLocalKeyMap map mapping remote keys to local key equivalents
@throws StoreException
ambry-replication/src/main/java/com/github/ambry/replication/ReplicaThread.java