static boolean acknowledge(MessageIdAdv msgId, boolean individual) {
        if (!isBatch(msgId)) {
            return true;
        }
        final BitSet ackSet = msgId.getAckSet();
        if (ackSet == null) {
            // The internal MessageId implementation should never reach here. If users have implemented their own
            // MessageId and getAckSet() is not override, return false to avoid acknowledge current entry.
            return false;
        }
        int batchIndex = msgId.getBatchIndex();
        synchronized (ackSet) {
            if (individual) {
                ackSet.clear(batchIndex);
            } else {
                ackSet.clear(0, batchIndex + 1);
            }
            return ackSet.isEmpty();
        }
    }
@Test
    public void testAcknowledgeIndividualConcurrently() throws InterruptedException {
        ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("pulsar-consumer-%d").build();
        @Cleanup("shutdown")
        ExecutorService executorService = Executors.newCachedThreadPool(threadFactory);
        for (int i = 0; i < 100; i++) {
            int batchSize = 32;
            BitSet bitSet = new BitSet(batchSize);
            bitSet.set(0, batchSize);
            AtomicInteger individualAcked = new AtomicInteger();
            Phaser phaser = new Phaser(1);
            CountDownLatch finishLatch = new CountDownLatch(batchSize);
            for (int batchIndex = 0; batchIndex < batchSize; batchIndex++) {
                phaser.register();
                BatchMessageIdImpl messageId = new BatchMessageIdImpl(1, 0, 0, batchIndex, batchSize, bitSet);
                executorService.execute(() -> {
                    try {
                        phaser.arriveAndAwaitAdvance();
                        if (MessageIdAdvUtils.acknowledge(messageId, true)) {
                            individualAcked.incrementAndGet();
                        }
                    } finally {
                        finishLatch.countDown();
                    }
                });
            }
            phaser.arriveAndDeregister();
            finishLatch.await();
            assertEquals(individualAcked.get(), 1);
        }
    }