@Override
  public void transfer(final TabletInsertionEvent tabletInsertionEvent) throws Exception {
    // Note: here we don't need to do type judgment here, because PipeConsensus uses
    // PIPE_CONSENSUS_PROCESSOR and will not change the event type like
    // org.apache.iotdb.db.pipe.connector.protocol.thrift.sync.IoTDBDataRegionSyncConnector
    try {
      if (isTabletBatchModeEnabled) {
        if (tabletBatchBuilder.onEvent(tabletInsertionEvent)) {
          doTransfer();
        }
      } else {
        final long startTime = System.nanoTime();
        doTransferWrapper((PipeInsertNodeTabletInsertionEvent) tabletInsertionEvent);
        final long duration = System.nanoTime() - startTime;
        pipeConsensusConnectorMetrics.recordRetryWALTransferTimer(duration);
      }
    } catch (final Exception e) {
      throw new PipeRuntimeConnectorRetryTimesConfigurableException(
          String.format(
              "Failed to transfer tablet insertion event %s, because %s.",
              tabletInsertionEvent, e.getMessage()),
          Integer.MAX_VALUE);
    }
  }
This connector is used for PipeConsensus to transfer queued event.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/pipeconsensus/PipeConsensusSyncConnector.java