[Bug] Java consumer occasionally missing one message of a batched entry

[YinY1]

Search before reporting

• I searched in the issues and found nothing similar.

Read release policy

• I understand that unsupported versions don't get bug fixes. I will attempt to reproduce the issue on a supported version of Pulsar client and Pulsar broker.

User environment

broker and client are in a same machine use same java

• OS: Linux version 5.4.241-1-tlinux4-0017.16 (mockbuild@VM-81-152-TS3) (gcc version 8.5.0 20210514 (Tencent 8.5.0-23) (GCC)) SMP Thu Dec 12 21:16:37 CST 2024
• Broker version: 4.1.1
• Client version: 4.1.1 (Java)
• Java version:
  • openjdk version "24.0.2" 2025-07-15
  • OpenJDK Runtime Environment Temurin-24.0.2+12 (build 24.0.2+12)
  • OpenJDK 64-Bit Server VM Temurin-24.0.2+12 (build 24.0.2+12, mixed mode, sharing)

Issue Description

What happened

What I did

Continuously produced a certain amount of messages to a partitioned topic, and consumed those messages with two subscriptions [sub-1, sub-2] individually, and check their consistency. Consistent means a message has been sent and received from all subscriptions, so it will be checked after received and before acknowledged.
options i used (use default if not mentioned)

• producer
  • batchingMaxMessagess = 1000
• consumer
  • enableBatchIndexAcknowledgment = true
  • ack mode = Individual
  • acknowledgmentGroupTime = 100ms
  • maxAcknowledgmentGroupSize = 1000

Expected

All messages produced could be received from all subscriptions.
If produced 1000000, I should get

--- Total Acked Messages per Subscription ---
2026-01-14 11:56:07.351 [main] INFO  org.example.Main - Subscription [sub-2]: 1000000 acks
2026-01-14 11:56:07.351 [main] INFO  org.example.Main - Subscription [sub-1]: 1000000 acks 

Happened

Occasionally, missed one message with a random batch index

--- Total Acked Messages per Subscription ---
2026-01-14 11:56:07.351 [main] INFO  org.example.Main - Subscription [sub-2]: 999999 acks
2026-01-14 11:56:07.351 [main] INFO  org.example.Main - Subscription [sub-1]: 1000000 acks 
2026-01-14 11:56:07.351 [main] ERROR org.example.Main - [0:123:0:45] not received from [sub-2]!

Why a bug

If set enableBatchIndexAcknowledgment = true or set acknowledgmentGroupTime = 0,
this will not occur.
And broker has a backlog pointed to these entry, while consumer.receive() could not get that missing message even after calling redeliverUnacknowledgedMessages()

** use go client (not support Batch Index Acknowledgment) will not occur **

Error messages

if enable DEBUG for .apache.pulsar.client.impl.PersistentAcknowledgmentsGroupingTracker, found

Flushing pending acks to broker: last-cumulative-ack: [] -- individual-acks: [] -- individual-batch-index-acks: [(0, 123, {})]

Reproducing the issue

pseudocode:

public class PulsarBatchAckPseudoDemo {

    static Set<MessageId> sentMessageIds = new ConcurrentHashMap.newKeySet();
    static Map<MessageId, Set<String>> receiptTracker = new ConcurrentHashMap<>();
    static Map<String, AtomicLong> ackCounters = Map.of("sub-1", new AtomicLong(0), "sub-2", new AtomicLong(0));

    public static void main(String[] args) {

        // Assume a partitioned topic is already created.
        PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://127.0.0.1:6650").build();

        // Run producer and consumers concurrently in background threads.
        startThread(() -> producerTask(client));
        startThread(() -> consumerTask(client, "sub-1"));
        startThread(() -> consumerTask(client, "sub-2"));

        // Let the test run for a while, then stop threads and verify.
        sleep(90_000); // Wait for a generous amount of time.

        // --- Verification ---
        System.out.println("--- Total Acked Messages per Subscription ---");
        System.out.printf("Subscription [sub-1]: %d acks%n", ackCounters.get("sub-1").get());
        System.out.printf("Subscription [sub-2]: %d acks%n", ackCounters.get("sub-2").get());

        // Find and report any message that wasn't received by BOTH subscriptions.
        for (MessageId sentId : sentMessageIds) {
            Set<String> receivedBy = receiptTracker.getOrDefault(sentId, Collections.emptySet());
            if (receivedBy.size() < 2) {
                if (!receivedBy.contains("sub-1")) {
                    System.err.printf("[%s] not received from [sub-1]!%n", sentId);
                }
                if (!receivedBy.contains("sub-2")) {
                    System.err.printf("[%s] not received from [sub-2]!%n", sentId);
                }
            }
        }
    }

    // --- Producer Task Logic ---
    @SneakyThrows
    static void producerTask(PulsarClient client) {
        Producer producer = client.newProducer().topic("partitioned_topic").batchingMaxMessages(1000).create();

        for (int i = 0; i < 1_000_000; i++) {
            // Asynchronously send and register the MessageId upon completion.
            producer.sendAsync("message-payload-" + i).thenAccept(messageId -> {
                sentMessageIds.add(messageId);
                receiptTracker.put(messageId, ConcurrentHashMap.newKeySet());
            });
        }
    }

    // --- Consumer Task Logic ---
    @SneakyThrows
    static void consumerTask(PulsarClient client, String subscriptionName) {
        Consumer consumer = client.newConsumer().topic("partitioned_topic").subscriptionName(subscriptionName)
                .enableBatchIndexAcknowledgement(true).acknowledgmentGroupTime(100, TimeUnit.MILLISECONDS)
                .acknowledgementGroupSize(1000).subscribe();

        while (true) { // Run until the main thread stops it.
            Message<String> message = consumer.receive();

            // Step 1: Mark as received BEFORE acknowledging.
            Set<String> receipts = receiptTracker.get(message.getMessageId());
            if (receipts != null) {
                receipts.add(subscriptionName);
            }

            // Step 2: Increment the counter for the final report.
            ackCounters.get(subscriptionName).incrementAndGet();

            // Step 3: Acknowledge the message individually.
            consumer.acknowledge(message);
        }
    }
}

Additional information

No response

Are you willing to submit a PR?

• I'm willing to submit a PR!

[lhotari]

I haven't checked it yet, but there's a possibility that the test code contains a race condition with the receiptTracker handling.
You can remove if (receipts != null) { in the consumer task logic to find out. It would be null if there's a race condition.

To fix the race condition, I'd recommend extracting a method for getting the receipt tracker and use something like receiveTracker.computeIfAbsent(messageId, __ -> ConcurrentHashMap.newKeySet()).

[YinY1]

I haven't checked it yet, but there's a possibility that the test code contains a race condition with the receiptTracker handling.
You can remove if (receipts != null) { in the consumer task logic to find out. It would be null if there's a race condition.

To fix the race condition, I'd recommend extracting a method for getting the receipt tracker and use something like receiveTracker.computeIfAbsent(messageId, __ -> ConcurrentHashMap.newKeySet()).

I'm sorry about my mistake in this demo. Because of the complexity of the the actual production code, I did not copy the code direcetly but rewrote a test demo without a careful check.

Indeed I do use computeIfAbsent and other thread safe methods in my code, I will correct this test demo ASAP.

This bug is so rare that during testing I produce and consume 100,000 messages per iteration; sometimes it can be reproduced within five test cycles, while other times it cannot be reproduced even after more than ten cycles.

Thanks for your patience again!

[lhotari]

@YinY1 I created https://github.com/lhotari/pulsar-playground/blob/master/src/main/java/com/github/lhotari/pulsar/playground/TestScenarioIssue25145.java based on PulsarBatchAckPseudoDemo. I wasn't able to reproduce the problem after fixing the race condition.

[YinY1]

@YinY1 I created https://github.com/lhotari/pulsar-playground/blob/master/src/main/java/com/github/lhotari/pulsar/playground/TestScenarioIssue25145.java based on PulsarBatchAckPseudoDemo. I wasn't able to reproduce the problem after fixing the race condition.

Thank you so much for providing the fix code! And I apologize for not addressing my code issue in a timely manner.

I used the code you provided and made some simple modifications so that I could run loop tests, and I modified the number of messages to produced in each iteration. The main changes are as follows:

public static void main(String[] args) throws PulsarClientException, InterruptedException, PulsarAdminException {
    // ...

                // Find and report any message that wasn't received by BOTH subscriptions.
+             boolean inconsistentFound = false;
                for (MessageId sentId : sentMessageIds) {
                    Set<String> receivedBy = receiptTracker.getOrDefault(sentId, Collections.emptySet());
                    if (receivedBy.size() < 2) {
  +                     inconsistentFound = true;
                        if (!receivedBy.contains("sub-1")) {
                            System.err.printf("[%s] not received from [sub-1]!%n", sentId);
                        }
                        if (!receivedBy.contains("sub-2")) {
                            System.err.printf("[%s] not received from [sub-2]!%n", sentId);
                        }
                    }
                }

                System.out.println("Shutting down...");
                client.close();
                System.out.println("Done.");
  + 
  +             if (inconsistentFound) {
  +                 System.err.println("Inconsistency detected!");
  +                 System.exit(1);
  +             }
            }

Additionally, I used a script to execute the program in a loop:

iteration=0
while true; do
        iteration=$((iteration + 1))
        echo "====== Iteration ${iteration} ======"
        echo "Start time: $(date)"
        rm -rf /workspace/output
        mkdir -p /workspace/output
        current_time=$(date +%Y%m%d-%H%M%S)

        java -jar /workspace/target/test-scenario-issue-25145-1.0-SNAPSHOT.jar 2>&1 | tee /workspace/log/output_${current_time}.log
        rc=${PIPESTATUS[0]}

        if [ "${rc}" -eq 0 ]; then
            echo "====== Iteration ${iteration} completed successfully ======"
            echo "Sleeping 10s before next iteration..."
            sleep 10
            continue
        else
            echo "====== Iteration ${iteration} failed with exit code ${rc} ======"
            echo "Inconsistency detected. Total iterations run: ${iteration}"
            echo "Log saved to: /workspace/log/output_${current_time}.log"
            exit ${rc}
        fi
done

After running for a while, the issue reappeared. Here is the output log:

====== Iteration 7 ======
Start time: Thu Jan 15 04:05:10 PM UTC 2026
SLF4J(W): No SLF4J providers were found.
SLF4J(W): Defaulting to no-operation (NOP) logger implementation
SLF4J(W): See https://www.slf4j.org/codes.html#noProviders for further details.
Starting producer task...
Starting consumer task for subscription [sub-2]...
Starting consumer task for subscription [sub-1]...
Finished producer task.
Finished consumer task for subscription [sub-1].
Finished consumer task for subscription [sub-2].
--- Total Acked Messages per Subscription ---
Subscription [sub-1]: 100000 acks
Subscription [sub-2]: 100000 acks
Shutting down...
Done.
====== Iteration 7 completed successfully ======
Sleeping 10s before next iteration...
====== Iteration 8 ======
Start time: Thu Jan 15 04:05:51 PM UTC 2026
SLF4J(W): No SLF4J providers were found.
SLF4J(W): Defaulting to no-operation (NOP) logger implementation
SLF4J(W): See https://www.slf4j.org/codes.html#noProviders for further details.
Starting producer task...
Starting consumer task for subscription [sub-1]...
Starting consumer task for subscription [sub-2]...
Finished producer task.
Finished consumer task for subscription [sub-2].
Finished consumer task for subscription [sub-1].
--- Total Acked Messages per Subscription ---
Subscription [sub-1]: 99999 acks
Subscription [sub-2]: 100000 acks
[5426:163:7:582] not received from [sub-1]!
Shutting down...
Done.
Inconsistency detected!
====== Iteration 8 failed with exit code 1 ======
Inconsistency detected. Total iterations run: 8

Unfortunately, I didn't enable the debug log for AckGroupingTracker, so I'm unable to provide the tracker's status at that time.

I've tested in different broker versions, (multi replicas, no standalone): 2.9.0, 4.0.8 and 4.1.1, and bug still occurred. As I mentioned above, I prefer this bug is caused by batchIndexAcknowledgement or PersistentAcknowledgmentsGroupingTracker.

Additionally, I run this test demo and broker 4.0.8 on k8s.

Thanks for your patience again!

[lhotari]

Additionally, I run this test demo and broker 4.0.8 on k8s.

@YinY1 Do you run the test case within the Pulsar cluster? Do you connect to brokers via the Pulsar Proxy?

[lhotari]

I've tried to reproduce the issue with 4.0.8 version in k8s without success. I'm running the test app outside of k8s.

I updated the test app to reflect similar changes as you have made:
https://github.com/lhotari/pulsar-playground/blob/master/src/main/java/com/github/lhotari/pulsar/playground/TestScenarioIssue25145.java
And I added the script and a k8s deployment:
https://github.com/lhotari/pulsar-playground/tree/master/issues/issue25145

[YinY1]

Additionally, I run this test demo and broker 4.0.8 on k8s.

@YinY1 Do you run the test case within the Pulsar cluster? Do you connect to brokers via the Pulsar Proxy?

I also run test app in k8s issue-25145.yaml, but not in the same pod/stateful set as Pulsar cluster.
For 4.0.8 version cluster (3 replicas), I used Pulsar Proxy, but for 4.1.1 version I connected directly without proxy.