ReplicaMetadataResponse getReplicaMetadataResponse(List<RemoteReplicaInfo> replicasToReplicatePerNode,
      ConnectedChannel connectedChannel, DataNodeId remoteNode) throws ReplicationException, IOException {
    long replicaMetadataRequestStartTime = time.milliseconds();
    ChannelOutput channelOutput = null;
    try {
      ReplicaMetadataRequest request = createReplicaMetadataRequest(replicasToReplicatePerNode, remoteNode);
      channelOutput = connectedChannel.sendAndReceive(request);
      logger.trace("Remote node: {} Thread name: {} Remote replicas: {} Stream size after deserialization: {} ",
          remoteNode, threadName, replicasToReplicatePerNode, channelOutput.getInputStream().available());
      ReplicaMetadataResponse response =
          ReplicaMetadataResponse.readFrom(channelOutput.getInputStream(), findTokenHelper, clusterMap);

      long metadataRequestTime = time.milliseconds() - replicaMetadataRequestStartTime;
      replicationMetrics.updateMetadataRequestTime(metadataRequestTime, replicatingFromRemoteColo, replicatingOverSsl,
          datacenterName);

      if (response.getError() != ServerErrorCode.NoError
          || response.getReplicaMetadataResponseInfoList().size() != replicasToReplicatePerNode.size()) {
        int replicaMetadataResponseInfoListSize = response.getReplicaMetadataResponseInfoList() == null ? 0
            : response.getReplicaMetadataResponseInfoList().size();
        logger.error(
            "Remote node: {} Thread name: {} Remote replicas: {} Replica metadata response error: {} ReplicaMetadataResponseInfoListSize: {} ReplicasToReplicatePerNodeSize: {}",
            remoteNode, threadName, replicasToReplicatePerNode, response.getError(),
            replicaMetadataResponseInfoListSize, replicasToReplicatePerNode.size());
        throw new ReplicationException("Replica Metadata Response Error " + response.getError());
      }
      return response;
    } catch (Exception e) {
      responseHandler.onEvent(replicasToReplicatePerNode.get(0).getReplicaId(), e);
      throw e;
    } finally {
      if (channelOutput != null && (channelOutput.getInputStream() instanceof NettyByteBufDataInputStream)) {
        // Release buffer if and only if the inputStream is NettyByteBuf based.
        ((NettyByteBufDataInputStream) channelOutput.getInputStream()).getBuffer().release();
      }
    }
  }
Gets the replica metadata response for a list of remote replicas on a given remote data node
@param replicasToReplicatePerNode The list of remote replicas for a node
@param connectedChannel The connection channel to the node
@param remoteNode The remote node from which replication needs to happen
@return ReplicaMetadataResponse, the response from replica metadata request to remote node
@throws ReplicationException
@throws IOException
ambry-replication/src/main/java/com/github/ambry/replication/ReplicaThread.java