@Override
  public void transfer(TsFileInsertionEvent tsFileInsertionEvent) throws Exception {
    asyncTransferQueuedEventsIfNecessary();
    transferBatchedEventsIfNecessary();

    if (!(tsFileInsertionEvent instanceof PipeTsFileInsertionEvent)) {
      LOGGER.warn(
          "PipeConsensusAsyncConnector only support PipeTsFileInsertionEvent. Current event: {}.",
          tsFileInsertionEvent);
      return;
    }

    boolean enqueueResult = addEvent2Buffer((EnrichedEvent) tsFileInsertionEvent);
    if (!enqueueResult) {
      throw new PipeRuntimeConnectorRetryTimesConfigurableException(
          ENQUEUE_EXCEPTION_MSG, Integer.MAX_VALUE);
    }

    transferWithoutCheck(tsFileInsertionEvent);
  }
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