public synchronized void removeEventFromBuffer(EnrichedEvent event) {
    if (LOGGER.isDebugEnabled()) {
      LOGGER.debug(
          "PipeConsensus-ConsensusGroup-{}: one event-{} successfully received by the follower, will be removed from queue, queue size = {}, limit size = {}",
          consensusGroupId,
          event,
          transferBuffer.size(),
          IOTDB_CONFIG.getIotConsensusV2PipelineSize());
    }
    if (transferBuffer.isEmpty()) {
      LOGGER.info(
          "PipeConsensus-ConsensusGroup-{}: try to remove event-{} after pipeConsensusAsyncConnector being closed. Ignore it.",
          consensusGroupId,
          event);
      return;
    }
    Iterator<EnrichedEvent> iterator = transferBuffer.iterator();
    EnrichedEvent current = iterator.next();
    while (!current.equalsInPipeConsensus(event) && iterator.hasNext()) {
      current = iterator.next();
    }
    iterator.remove();
    // update replicate progress
    currentReplicateProgress =
        Math.max(currentReplicateProgress, event.getReplicateIndexForIoTV2());
    // decrease reference count
    event.decreaseReferenceCount(PipeConsensusAsyncConnector.class.getName(), true);
  }
if one event is successfully processed by receiver in PipeConsensus, we will remove this event
from transferBuffer in order to transfer other event.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/pipeconsensus/PipeConsensusAsyncConnector.java