private boolean addEvent2Buffer(EnrichedEvent event) {
    try {
      if (LOGGER.isDebugEnabled()) {
        LOGGER.debug(
            "PipeConsensus-ConsensusGroup-{}: no.{} event-{} added to connector buffer",
            consensusGroupId,
            event.getReplicateIndexForIoTV2(),
            event);
      }
      // Special judge to avoid transfer stuck when re-transfer events that will not be put in
      // retryQueue.
      if (transferBuffer.contains(event)) {
        return true;
      }
      long currentTime = System.nanoTime();
      boolean result =
          transferBuffer.offer(
              event, PIPE_CONSENSUS_EVENT_ENQUEUE_TIMEOUT_IN_MS, TimeUnit.MILLISECONDS);
      long duration = System.nanoTime() - currentTime;
      pipeConsensusConnectorMetrics.recordConnectorEnqueueTimer(duration);
      // add reference
      if (result) {
        event.increaseReferenceCount(PipeConsensusAsyncConnector.class.getName());
      }
      // if connector is closed when executing this method, need to clear this event's reference
      // count to avoid unnecessarily pinning some resource such as WAL.
      if (isClosed.get()) {
        event.clearReferenceCount(PipeConsensusAsyncConnector.class.getName());
      }
      return result;
    } catch (InterruptedException e) {
      LOGGER.info("PipeConsensusConnector transferBuffer queue offer is interrupted.", e);
      Thread.currentThread().interrupt();
      return false;
    }
  }
Add an event to transferBuffer, whose events will be asynchronously transferred to receiver.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/pipeconsensus/PipeConsensusAsyncConnector.java