Set<MessageInfo> getMissingStoreMessages(ReplicaMetadataResponseInfo replicaMetadataResponseInfo,
      DataNodeId remoteNode, RemoteReplicaInfo remoteReplicaInfo) throws StoreException {
    long startTime = time.milliseconds();
    List<MessageInfo> messageInfoList = replicaMetadataResponseInfo.getMessageInfoList();
    Map<MessageInfo, StoreKey> remoteMessageToConvertedKeyNonNull = new HashMap<>();
    for (MessageInfo messageInfo : messageInfoList) {
      StoreKey storeKey = messageInfo.getStoreKey();
      logger.trace("Remote node: {} Thread name: {} Remote replica: {} Key from remote: {}", remoteNode, threadName,
          remoteReplicaInfo.getReplicaId(), storeKey);
      StoreKey convertedKey = storeKeyConverter.getConverted(storeKey);
      if (skipPredicate == null) {
        logger.debug("SkipPredicate is null");
      }
      if (convertedKey != null && (!replicationConfig.replicationContainerDeletionEnabled || skipPredicate == null
          || !skipPredicate.test(messageInfo))) {
        remoteMessageToConvertedKeyNonNull.put(messageInfo, convertedKey);
      }
    }
    Set<StoreKey> convertedMissingStoreKeys =
        remoteReplicaInfo.getLocalStore().findMissingKeys(new ArrayList<>(remoteMessageToConvertedKeyNonNull.values()));
    Set<MessageInfo> missingRemoteMessages = new HashSet<>();
    remoteMessageToConvertedKeyNonNull.forEach((messageInfo, convertedKey) -> {
      if (convertedMissingStoreKeys.contains(convertedKey)) {
        logger.trace(
            "Remote node: {} Thread name: {} Remote replica: {} Key missing id (converted): {} Key missing id (remote): {}",
            remoteNode, threadName, remoteReplicaInfo.getReplicaId(), convertedKey, messageInfo.getStoreKey());
        missingRemoteMessages.add(messageInfo);
      }
    });
    if (messageInfoList.size() != 0 && missingRemoteMessages.size() == 0) {
      // Catching up
      replicationMetrics.allResponsedKeysExist.inc();
    }
    replicationMetrics.updateCheckMissingKeysTime(time.milliseconds() - startTime, replicatingFromRemoteColo,
        datacenterName);
    return missingRemoteMessages;
  }
Gets the missing store messages by comparing the messages from the remote node
@param replicaMetadataResponseInfo The response that contains the messages from the remote node
@param remoteNode The remote node from which replication needs to happen
@param remoteReplicaInfo The remote replica that contains information about the remote replica id
@return List of store messages that are missing from the local store
@throws StoreException if store error (usually IOError) occurs when getting missing keys.
ambry-replication/src/main/java/com/github/ambry/replication/ReplicaThread.java