private boolean ackInternal(
      final String consumerId, final SubscriptionCommitContext commitContext) {
    final AtomicBoolean acked = 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;
          }

          if (ev.isCommitted()) {
            LOGGER.warn(
                "Subscription: subscription event {} is committed, subscription commit context {}, prefetching queue: {}",
                ev,
                commitContext,
                this);
            // clean up committed event
            ev.cleanUp(false);
            return null; // remove this entry
          }

          if (!ev.isCommittable()) {
            LOGGER.warn(
                "Subscription: subscription event {} is not committable, subscription commit context {}, prefetching queue: {}",
                ev,
                commitContext,
                this);
            return ev;
          }

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

          ev.ack();
          ev.recordCommittedTimestamp(); // now committed
          acked.set(true);

          // clean up committed event
          ev.cleanUp(false);
          return null; // remove this entry
        });

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