private PipeConsensusReceiver internalSetAndGetReceiver(
      ConsensusGroupId consensusGroupId,
      ConsensusPipeName consensusPipeName,
      byte reqVersion,
      AtomicBoolean isFirstGetReceiver) {
    // 1. Route to given consensusGroup's receiver map
    Map<ConsensusPipeName, AtomicReference<PipeConsensusReceiver>> consensusPipe2ReciverMap =
        replicaReceiverMap.get(consensusGroupId);
    // 2. Route to given consensusPipeTask's receiver
    AtomicReference<PipeConsensusReceiver> receiverReference =
        consensusPipe2ReciverMap.get(consensusPipeName);

    if (isFirstGetReceiver.get()) {
      if (RECEIVER_CONSTRUCTORS.containsKey(reqVersion)) {
        // If is first get receiver, set this receiver as alive.
        aliveReceivers.add(consensusPipeName);
        receiverReference.set(
            RECEIVER_CONSTRUCTORS
                .get(reqVersion)
                .apply(pipeConsensus, consensusGroupId, consensusPipeName));
        LOGGER.info("Receiver-{} is ready", consensusPipeName);
      } else {
        throw new UnsupportedOperationException(
            String.format("Unsupported pipeConsensus request version %d", reqVersion));
      }
    } else {
      waitUntilReceiverGetInitiated(receiverReference);
    }
    return receiverReference.get();
  }
For each consensus Pipe task, there is an independent receiver. So for every replica, it has
(n-1) receivers, n is the num of replicas. 1 DataNode --has--> 1 PipeConsensusReceiverAgent & n
replicas 1 PipeConsensusReceiverAgent --manages--> n replicas' receivers 1 replica --has-->
(n-1) receivers
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/pipeconsensus/PipeConsensusReceiverAgent.java