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

    final long requestCommitId = ((EnrichedEvent) event).getReplicateIndexForIoTV2();

    // The deduplication logic here is to avoid the accumulation of the same event in a batch when
    // retrying.
    if ((events.isEmpty() || !events.get(events.size() - 1).equals(event))) {
      events.add((EnrichedEvent) event);
      requestCommitIds.add(requestCommitId);
      final int bufferSize = buildTabletInsertionBuffer(event);

      ((EnrichedEvent) event)
          .increaseReferenceCount(PipeConsensusTransferBatchReqBuilder.class.getName());

      if (firstEventProcessingTime == Long.MIN_VALUE) {
        firstEventProcessingTime = System.currentTimeMillis();
      }

      totalBufferSize += bufferSize;
    }

    return totalBufferSize >= getMaxBatchSizeInBytes()
        || System.currentTimeMillis() - firstEventProcessingTime >= maxDelayInMs;
  }
Try offer {@link Event} into cache if the given {@link Event} is not duplicated.

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