@Override
  public final void releaseReceiverResource(DataRegionId dataRegionId) {
    // 1. Route to given consensusGroup's receiver map
    Map<ConsensusPipeName, AtomicReference<PipeConsensusReceiver>> consensusPipe2ReciverMap =
        this.replicaReceiverMap.getOrDefault(
            ConsensusGroupId.Factory.create(
                TConsensusGroupType.DataRegion.getValue(), dataRegionId.getId()),
            new ConcurrentHashMap<>());
    // 2. Release all related receivers
    consensusPipe2ReciverMap.entrySet().stream()
        .filter(entry -> entry.getKey().getReceiverDataNodeId() == thisNodeId)
        .forEach(
            receiverEntry -> {
              ConsensusPipeName consensusPipeName = receiverEntry.getKey();
              AtomicReference<PipeConsensusReceiver> receiverReference = receiverEntry.getValue();
              if (receiverReference != null) {
                // no longer receive new request
                aliveReceivers.remove(consensusPipeName);
                receiverReference.get().handleExit();
                receiverReference.set(null);
              }
            });
    // 3. Release replica map
    this.replicaReceiverMap.remove(dataRegionId);
    // 4. GC receiver map
    consensusPipe2ReciverMap.clear();
    LOGGER.info("All Receivers related to {} are released.", dataRegionId);
  }
Release all receivers of given data region
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/pipeconsensus/PipeConsensusReceiverAgent.java