public synchronized void onEvent(final TabletInsertionEvent event)
      throws IOException, WALPipeException {
    if (!(event instanceof EnrichedEvent)) {
      LOGGER.warn(
          "Unsupported event {} type {} when building transfer request", event, event.getClass());
      return;
    }

    if (!useLeaderCache) {
      defaultBatch.onEvent(event);
      return;
    }

    String deviceId = null;
    if (event instanceof PipeRawTabletInsertionEvent) {
      deviceId = ((PipeRawTabletInsertionEvent) event).getDeviceId();
    } else if (event instanceof PipeInsertNodeTabletInsertionEvent) {
      deviceId = ((PipeInsertNodeTabletInsertionEvent) event).getDeviceId();
    }

    if (Objects.isNull(deviceId)) {
      defaultBatch.onEvent(event);
      return;
    }

    final TEndPoint endPoint =
        IoTDBDataNodeCacheLeaderClientManager.LEADER_CACHE_MANAGER.getLeaderEndPoint(deviceId);
    if (Objects.isNull(endPoint)) {
      defaultBatch.onEvent(event);
      return;
    }
    endPointToBatch
        .computeIfAbsent(
            endPoint,
            k -> new PipeTabletEventPlainBatch(requestMaxDelayInMs, requestMaxBatchSizeInBytes))
        .onEvent(event);
  }
Try offer {@link Event} into the corresponding batch if the given {@link Event} is not
duplicated.

@param event the given {@link Event}
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/payload/evolvable/batch/PipeTransferBatchReqBuilder.java