private void handleGetResponse(GetResponse getResponse, List<RemoteReplicaInfo> replicasToReplicatePerNode,
      List<ExchangeMetadataResponse> exchangeMetadataResponseList, DataNodeId remoteNode,
      boolean remoteColoGetRequestForStandby) throws IOException {
    int partitionResponseInfoIndex = 0;
    long totalBytesFixed = 0;
    long totalBlobsFixed = 0;
    long startTime = time.milliseconds();
    for (int i = 0; i < exchangeMetadataResponseList.size(); i++) {
      ExchangeMetadataResponse exchangeMetadataResponse = exchangeMetadataResponseList.get(i);
      RemoteReplicaInfo remoteReplicaInfo = replicasToReplicatePerNode.get(i);
      // TODO: if remoteReplicaInfo.getLocalStore() is closed, write will fail
      if (exchangeMetadataResponse.serverErrorCode == ServerErrorCode.NoError) {
        if (exchangeMetadataResponse.missingStoreMessages.size() > 0) {
          PartitionResponseInfo partitionResponseInfo =
              getResponse.getPartitionResponseInfoList().get(partitionResponseInfoIndex);
          responseHandler.onEvent(remoteReplicaInfo.getReplicaId(), partitionResponseInfo.getErrorCode());
          partitionResponseInfoIndex++;
          if (!partitionResponseInfo.getPartition()
              .toPathString()
              .equals(remoteReplicaInfo.getReplicaId().getPartitionId().toPathString())) {
            throw new IllegalStateException(
                "The partition id from partitionResponseInfo " + partitionResponseInfo.getPartition()
                    + " and from remoteReplicaInfo " + remoteReplicaInfo.getReplicaId().getPartitionId()
                    + " are not the same");
          }
          if (partitionResponseInfo.getErrorCode() == ServerErrorCode.NoError) {
            List<MessageInfo> messageInfoList = partitionResponseInfo.getMessageInfoList();
            try {
              logger.trace("Remote node: {} Thread name: {} Remote replica: {} Messages to fix: {} "
                      + "Partition: {} Local mount path: {}", remoteNode, threadName, remoteReplicaInfo.getReplicaId(),
                  exchangeMetadataResponse.getMissingStoreKeys(), remoteReplicaInfo.getReplicaId().getPartitionId(),
                  remoteReplicaInfo.getLocalReplicaId().getMountPath());

              List<Transformer> transformers = transformer == null ? Collections.emptyList()
                  : Collections.singletonList(transformer);
              MessageSievingInputStream validMessageDetectionInputStream =
                  new MessageSievingInputStream(getResponse.getInputStream(), messageInfoList,
                      transformers, metricRegistry);
              String replicaName = String.format("%s:%s", remoteReplicaInfo.getReplicaId().getDataNodeId().getHostname(),
                  remoteReplicaInfo.getReplicaId().getReplicaPath());
              validMessageDetectionInputStream.setReplicaLocation(replicaName);
              if (validMessageDetectionInputStream.hasInvalidMessages()) {
                replicationMetrics.incrementInvalidMessageError(partitionResponseInfo.getPartition());
                logger.error("Out of {} messages, {} invalid messages were found in message stream from {}",
                    messageInfoList.size(),
                    messageInfoList.size() - validMessageDetectionInputStream.getValidMessageInfoList().size(),
                    remoteReplicaInfo.getReplicaId());
              }
              messageInfoList = validMessageDetectionInputStream.getValidMessageInfoList();
              applyPut(validMessageDetectionInputStream, remoteReplicaInfo);

              for (MessageInfo messageInfo : messageInfoList) {
                totalBytesFixed += messageInfo.getSize();

                replicationMetrics.updateReplicationFetchBytes(remoteReplicaInfo, messageInfo.getSize());
                logger.trace("Remote node: {} Thread name: {} Remote replica: {} Message replicated: {} Partition: {} "
                        + "Local mount path: {} Message size: {}", remoteNode, threadName, remoteReplicaInfo.getReplicaId(),
                    messageInfo.getStoreKey(), remoteReplicaInfo.getReplicaId().getPartitionId(),
                    remoteReplicaInfo.getLocalReplicaId().getMountPath(), messageInfo.getSize());
                if (notification != null) {
                  notification.onBlobReplicaCreated(dataNodeId.getHostname(), dataNodeId.getPort(),
                      messageInfo.getStoreKey().getID(), BlobReplicaSourceType.REPAIRED);
                }
                // messageInfo is the Blob final state and the operation time is the blob creation time.
                // So the applyTtlUpdate will use the creation time as the Ttl operation time.
                if (isTtlUpdateNeededAfterPut(messageInfo)) {
                  applyTtlUpdate(messageInfo, remoteReplicaInfo);
                }
                StoreKey key = messageInfo.getStoreKey();
                if (key instanceof BlobId) {
                  BlobId blobId = (BlobId) key;
                  // Replication Lag should be recorded against the originating datacenter
                  replicationMetrics.updateReplicationLagInSecondsForBlob(
                      clusterMap.getDatacenterName(blobId.getDatacenterId()),
                      (time.milliseconds() - messageInfo.getOperationTimeMs()) / Time.MsPerSec);
                }
              }
              totalBlobsFixed += messageInfoList.size();

              if (leaderBasedReplicationAdmin != null) {
                // If leader based replication is enabled, we will only fetch missing blobs for local leaders from their
                // remote leaders. For non-leader replicas pairs (leader <-> standby, standby <-> leader, standby <->
                // standby), we will store the missing keys and track them via leader<->leader exchanges and intra-dc
                // replication.
                // Notify all the replicas of the partition on newly written messages so that non-leader replica pairs
                // can update their missing keys and advance token if needed.
                leaderBasedReplicationAdmin.onMessageWriteForPartition(partitionResponseInfo.getPartition(),
                    messageInfoList);
              }

              if (validMessageDetectionInputStream.hasInvalidMessages()
                  && !remoteReplicaInfo.isReplicationRetryCountMaxed()) {
                // If we have a few retries left, then increment the retry count and retry replication
                long retryCount = remoteReplicaInfo.incReplicationRetryCount();
                logger.error(
                    "Retrying replication due to {} corrupted messages. thread={}, replicationRetryCount={}, token={}, lagInBytes={}, partition={}, replicaId={}",
                    validMessageDetectionInputStream.getNumInvalidMessages(), threadName, retryCount,
                    remoteReplicaInfo.getToken(), remoteReplicaInfo.getLocalLagFromRemoteInBytes(),
                    remoteReplicaInfo.getReplicaId().getPartitionId().toString(), remoteReplicaInfo.getReplicaId());
              } else {
                if (validMessageDetectionInputStream.hasInvalidMessages()) {
                  replicationMetrics.replicationNumBlobsSkippedAfterRetry.inc(
                      validMessageDetectionInputStream.getNumInvalidMessages());
                  logger.error(
                      "Unable to replicate {} corrupted messages. thread={}, replicationRetryCount={}, token={}, lagInBytes={}, partition={}, replicaId={}",
                      validMessageDetectionInputStream.getNumInvalidMessages(), threadName,
                      remoteReplicaInfo.getReplicationRetryCount(), remoteReplicaInfo.getToken(),
                      remoteReplicaInfo.getLocalLagFromRemoteInBytes(),
                      remoteReplicaInfo.getReplicaId().getPartitionId().toString(), remoteReplicaInfo.getReplicaId());
                  logToExternalTable(validMessageDetectionInputStream.getInvalidMessages(), remoteReplicaInfo);
                }
                // If there are no invalid messages, or we have maxed out the retries, then advance token
                advanceToken(remoteReplicaInfo, exchangeMetadataResponse);
              }

              logger.trace("Remote node: {} Thread name: {} Remote replica: {} Token after speaking to remote node: {}",
                  remoteNode, threadName, remoteReplicaInfo.getReplicaId(), exchangeMetadataResponse.remoteToken);
            } catch (StoreException e) {
              if (e.getErrorCode() != StoreErrorCodes.AlreadyExist) {
                replicationMetrics.updateLocalStoreError(remoteReplicaInfo.getReplicaId());
                logger.error("Remote node: {} Thread name: {} Remote replica: {}", remoteNode, threadName,
                    remoteReplicaInfo.getReplicaId(), e);
              }
            }
          } else if (partitionResponseInfo.getErrorCode() == ServerErrorCode.BlobDeleted) {
            replicationMetrics.blobDeletedOnGetCount.inc();
            logger.trace("One of the blobs to GET is deleted: Remote node: {} Thread name: {} Remote replica: {}",
                remoteNode, threadName, remoteReplicaInfo.getReplicaId());
          } else if (partitionResponseInfo.getErrorCode() == ServerErrorCode.BlobAuthorizationFailure) {
            replicationMetrics.blobAuthorizationFailureCount.inc();
            logger.error(
                "One of the blobs authorization failed: Remote node: {} Thread name: {} Remote replica: {} Keys are: {}",
                remoteNode, threadName, remoteReplicaInfo.getReplicaId(),
                exchangeMetadataResponse.getMissingStoreKeys());
          } else {
            replicationMetrics.incrementReplicationErrorCount(replicatingFromRemoteColo, datacenterName);
            replicationMetrics.updateGetRequestError(remoteReplicaInfo.getReplicaId());
            logger.error("Remote node: {} Thread name: {} Remote replica: {} Server get-response error: {}",
                remoteNode, threadName, remoteReplicaInfo.getReplicaId(), partitionResponseInfo.getErrorCode());
          }
        }
      }
    }
    long batchStoreWriteTime = time.milliseconds() - startTime;
    replicationMetrics.updateBatchStoreWriteTime(batchStoreWriteTime, totalBytesFixed, totalBlobsFixed,
        replicatingFromRemoteColo, replicatingOverSsl, datacenterName, remoteColoGetRequestForStandby);
  }
Handle the {@link GetResponse} from the remote data node, it will write the messages (if any) to the local stores
from the remote stores for the missing keys, and advances tokens.
@param getResponse The {@link GetResponse} that contains the missing messages. This may be null if there are no
                   missing messages to write as per the exchange metadata response. In that case this method will
                   simply advance the tokens for every store.
@param replicasToReplicatePerNode The list of remote replicas for the remote node
@param exchangeMetadataResponseList The list of metadata response from the remote node
@param remoteNode The remote node from which replication needs to happen
@param remoteColoGetRequestForStandby boolean which indicates if we are getting missing keys for standby or
                                      non-leader replica pairs during leader-based replication.
@throws IOException
ambry-replication/src/main/java/com/github/ambry/replication/ReplicaThread.java