public boolean nackInternal(
      final String consumerId, final SubscriptionCommitContext commitContext) {
    final AtomicBoolean nacked = new AtomicBoolean(false);
    inFlightEvents.compute(
        new Pair<>(consumerId, commitContext),
        (key, ev) -> {
          if (Objects.isNull(ev)) {
            LOGGER.warn(
                "Subscription: subscription commit context [{}] does not exist, it may have been committed or something unexpected happened, prefetching queue: {}",
                commitContext,
                this);
            return null;
          }

          // check if a consumer nacks event from another consumer group...
          final String consumerGroupId = commitContext.getConsumerGroupId();
          if (!Objects.equals(consumerGroupId, brokerId)) {
            LOGGER.warn(
                "inconsistent consumer group when nacking event, current: {}, incoming: {}, consumer id: {}, event commit context: {}, prefetching queue: {}, commit it anyway...",
                brokerId,
                consumerGroupId,
                consumerId,
                commitContext,
                this);
          }

          ev.nack(); // now pollable
          nacked.set(true);

          // no need to update inFlightEvents and prefetchingQueue
          return ev;
        });

    return nacked.get();
  }
@return {@code true} if nack successfully
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingQueue.java