public SubscriptionEvent pollInternal(final String consumerId) {
    states.markPollRequest();

    if (prefetchingQueue.isEmpty()) {
      states.markMissingPrefetch();
      try {
        executeReceiverSubtask(
            () -> {
              tryPrefetch();
              return null;
            },
            SubscriptionAgent.receiver().remainingMs());
      } catch (final Exception e) {
        LOGGER.warn("Exception {} occurred when {} execute receiver subtask", this, e, e);
      }
    }

    if (prefetchingQueue.isEmpty()) {
      onEvent();
    }

    final long size = prefetchingQueue.size();
    long count = 0;

    SubscriptionEvent event;
    try {
      while (count++ < size // limit control
          && Objects.nonNull(
              event =
                  prefetchingQueue.poll(
                      SubscriptionConfig.getInstance().getSubscriptionPollMaxBlockingTimeMs(),
                      TimeUnit.MILLISECONDS))) {
        if (event.isCommitted()) {
          LOGGER.warn(
              "Subscription: SubscriptionPrefetchingQueue {} poll committed event {} from prefetching queue (broken invariant), remove it",
              this,
              event);
          // no need to update inFlightEvents
          continue;
        }

        if (!event.pollable()) {
          LOGGER.warn(
              "Subscription: SubscriptionPrefetchingQueue {} poll non-pollable event {} from prefetching queue (broken invariant), nack and remove it",
              this,
              event);
          event.nack(); // now pollable
          // no need to update inFlightEvents and prefetchingQueue
          continue;
        }

        // This operation should be performed before updating inFlightEvents to prevent multiple
        // consumers from consuming the same event.
        event.recordLastPolledTimestamp(); // now non-pollable

        inFlightEvents.put(new Pair<>(consumerId, event.getCommitContext()), event);
        event.recordLastPolledConsumerId(consumerId);
        return event;
      }
    } catch (final InterruptedException e) {
      Thread.currentThread().interrupt();
      LOGGER.warn(
          "Subscription: SubscriptionPrefetchingQueue {} interrupted while polling events.",
          this,
          e);
    }

    return null;
  }
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