private synchronized void peekOnce() {
    final Event peekedEvent = UserDefinedEnrichedEvent.maybeOf(inputPendingQueue.peek());
    if (Objects.isNull(peekedEvent)) {
      return;
    }

    if (!(peekedEvent instanceof PipeHeartbeatEvent)) {
      return;
    }

    final Event polledEvent = inputPendingQueue.waitedPoll();
    if (!Objects.equals(peekedEvent, polledEvent)) {
      LOGGER.warn(
          "Subscription: inconsistent heartbeat event when {} peeking (broken invariant), expected {}, actual {}, offer back",
          this,
          peekedEvent,
          polledEvent);
      inputPendingQueue.directOffer(polledEvent);
    } else {
      ((PipeHeartbeatEvent) peekedEvent)
          .decreaseReferenceCount(SubscriptionPrefetchingQueue.class.getName(), false);
    }
  }
A ReentrantReadWriteLock to ensure thread-safe operations. This lock is used to guarantee
mutual exclusion between the {@link SubscriptionPrefetchingQueue#cleanUp} operation and other
operations such as {@link SubscriptionPrefetchingQueue#poll}, {@link
SubscriptionPrefetchingQueue#ack}, etc. However, it does not enforce mutual exclusion among the
other operations themselves.

<p>Under the premise of obtaining this lock, to avoid inconsistencies, updates to the {@link
SubscriptionEvent} in both {@link SubscriptionPrefetchingQueue#prefetchingQueue} and {@link
SubscriptionPrefetchingQueue#inFlightEvents} MUST be performed within the {@link
ConcurrentHashMap#compute} method of inFlightEvents in the {@link
SubscriptionPrefetchingQueue}.

<p>This lock is created with fairness set to true, which means threads acquire the lock in the
order they requested it, to avoid thread starvation.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingQueue.java