public synchronized boolean onEvent(final TabletInsertionEvent event)
      throws WALPipeException, IOException {
    if (isClosed || !(event instanceof EnrichedEvent)) {
      return false;
    }

    // The deduplication logic here is to avoid the accumulation of
    // the same event in a batch when retrying.
    if (events.isEmpty() || !Objects.equals(events.get(events.size() - 1), event)) {
      // We increase the reference count for this event to determine if the event may be released.
      if (((EnrichedEvent) event)
          .increaseReferenceCount(PipeTransferBatchReqBuilder.class.getName())) {

        try {
          if (constructBatch(event)) {
            events.add((EnrichedEvent) event);
          }
        } catch (final Exception e) {
          // If the event is not added to the batch, we need to decrease the reference count.
          ((EnrichedEvent) event)
              .decreaseReferenceCount(PipeTransferBatchReqBuilder.class.getName(), false);
          // Will cause a retry
          throw e;
        }

        if (firstEventProcessingTime == Long.MIN_VALUE) {
          firstEventProcessingTime = System.currentTimeMillis();
        }
      } else {
        LOGGER.warn("Cannot increase reference count for event: {}, ignore it in batch.", event);
      }
    }

    return shouldEmit();
  }
Try offer {@link Event} into batch if the given {@link Event} is not duplicated.

@param event the given {@link Event}
@return {@code true} if the batch can be transferred
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/payload/evolvable/batch/PipeTabletEventBatch.java