@Override
  public Event waitedPoll() {
    Event event = null;
    final int pollHistoricalTsFileThreshold =
        PIPE_CONFIG.getPipeRealTimeQueuePollHistoricalTsFileThreshold();
    final int realTimeQueueMaxWaitingTsFileSize =
        PIPE_CONFIG.getPipeRealTimeQueueMaxWaitingTsFileSize();

    if (pollTsFileCounter.get() >= PIPE_CONFIG.getPipeRealTimeQueuePollTsFileThreshold()
        && offerTsFileCounter.get() < realTimeQueueMaxWaitingTsFileSize) {
      event =
          pollHistoricalTsFileCounter.incrementAndGet() % pollHistoricalTsFileThreshold == 0
              ? tsfileInsertEventDeque.pollFirst()
              : tsfileInsertEventDeque.pollLast();
      pollTsFileCounter.set(0);
    }
    if (event == null) {
      // Sequentially poll the first offered non-TsFileInsertionEvent
      event = super.directPoll();
      if (event == null && !tsfileInsertEventDeque.isEmpty()) {
        event =
            pollHistoricalTsFileCounter.incrementAndGet() % pollHistoricalTsFileThreshold == 0
                ? tsfileInsertEventDeque.pollFirst()
                : tsfileInsertEventDeque.pollLast();
      }
      if (event != null) {
        pollTsFileCounter.incrementAndGet();
      }
    }

    // If no event is available, block until an event is available
    if (Objects.isNull(event) && offerTsFileCounter.get() < realTimeQueueMaxWaitingTsFileSize) {
      event = super.waitedPoll();
      if (Objects.isNull(event)) {
        event =
            pollHistoricalTsFileCounter.incrementAndGet() % pollHistoricalTsFileThreshold == 0
                ? tsfileInsertEventDeque.pollFirst()
                : tsfileInsertEventDeque.pollLast();
      }
      if (event != null) {
        pollTsFileCounter.incrementAndGet();
      }
    }

    return event;
  }
When the number of polls exceeds the pollHistoryThreshold, the {@link TsFileInsertionEvent} of
the earliest write to the queue is returned. if the pollHistoryThreshold is not reached then an
attempt is made to poll the queue for the latest insertion {@link Event}. First, it tries to
poll the first provided If there is no such {@link Event}, poll the last supplied {@link
TsFileInsertionEvent}. If no {@link Event} is available, it blocks until a {@link Event} is
available.

@return the freshest insertion {@link Event}. can be {@code null} if no {@link Event} is
    available.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/connector/PipeRealtimePriorityBlockingQueue.java