private PipeConsensusReceiver getReceiver(
      ConsensusGroupId consensusGroupId, int leaderDataNodeId, byte reqVersion) {
    // 1. Route to given consensusGroup's receiver map
    Map<ConsensusPipeName, AtomicReference<PipeConsensusReceiver>> consensusPipe2ReceiverMap =
        replicaReceiverMap.computeIfAbsent(consensusGroupId, key -> new ConcurrentHashMap<>());
    // 2. Route to given consensusPipeTask's receiver
    ConsensusPipeName consensusPipeName =
        new ConsensusPipeName(consensusGroupId, leaderDataNodeId, thisNodeId);
    AtomicBoolean isFirstGetReceiver = new AtomicBoolean(false);
    AtomicReference<PipeConsensusReceiver> receiverReference =
        consensusPipe2ReceiverMap.computeIfAbsent(
            consensusPipeName,
            key -> {
              isFirstGetReceiver.set(true);
              return new AtomicReference<>(null);
            });

    // 3. If not first get receiver && receiver is not alive, return null.
    if (!isFirstGetReceiver.get() && !aliveReceivers.contains(consensusPipeName)) {
      return null;
    }

    if (receiverReference.get() == null) {
      return internalSetAndGetReceiver(
          consensusGroupId, consensusPipeName, reqVersion, isFirstGetReceiver);
    }

    final byte receiverThreadLocalVersion = receiverReference.get().getVersion().getVersion();
    if (receiverThreadLocalVersion != reqVersion) {
      LOGGER.warn(
          "The pipeConsensus request version {} is different from the sender request version {},"
              + " the receiver will be reset to the sender request version.",
          receiverThreadLocalVersion,
          reqVersion);
      receiverReference.set(null);
      return internalSetAndGetReceiver(
          consensusGroupId, consensusPipeName, reqVersion, isFirstGetReceiver);
    }
    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