Shawyeok opened a new issue, #26125: URL: https://github.com/apache/pulsar/issues/26125
### Search before reporting - [x] I searched in the [issues](https://github.com/apache/pulsar/issues) and found nothing similar. ### Read release policy - [x] I understand that [unsupported versions](https://pulsar.apache.org/contribute/release-policy/#supported-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 version: reproduced on 3.0.17, 4.0.11, and current `master` - Broker OS: Linux (Docker `apachepulsar/pulsar` images) and macOS (source build, standalone) - Client library type: Java - Client library version: 3.0.17, 4.0.11 (matching broker version under test) - Deployment: standalone Note: batch-index acknowledgment (`acknowledgmentAtBatchIndexLevelEnabled`) has been enabled by default since Pulsar 4.0 (PIP-391), so no special broker configuration is required to hit this on 4.0+ — it only needed to be set explicitly to reproduce on 3.0.17. ### Issue Description With a `Shared` subscription, batch-index acknowledgment enabled, and a `DeadLetterPolicy` configured, the consumer can send a batch message to the DLQ topic even though the application already explicitly acknowledged that exact message on the **final** redelivery round (i.e. at `redeliveryCount == maxRedeliverCount`, which should be the last chance to prevent DLQ routing). - What I was trying to do: verify that acking a message at the last allowed redelivery round prevents it from being sent to the DLQ. - What I expected: a message the application acknowledges at/before `redeliveryCount == maxRedeliverCount` is never published to the DLQ topic. - What actually happened: two messages that were explicitly acked at `redeliveryCount == maxRedeliverCount` were still delivered to the DLQ topic. I believe this is a bug because the broker/client both end up considering the subscription fully acked (`pulsar-admin topics stats` shows `unackedMessages: 0`, `backlogSize: 0` once the test settles), yet the DLQ topic already received copies of messages the application successfully acknowledged — so those messages are effectively (and silently) duplicated into the DLQ with no way for the application to prevent it. Not yet root-caused. Working hypothesis (unverified): a race on the client between the `ackTimeout`/redelivery-count-driven "route to DLQ" decision (likely in `ConsumerBase`/`UnAckedMessageTracker`) and the application's own `acknowledge()` call, such that a message reaching `redeliveryCount == maxRedeliverCount` can be routed to the DLQ producer concurrently with the application acking it in that same round. ### Error messages No exceptions/stack traces — this is a silent logic/race bug, not a crash. ### Reproducing the issue Start a standalone broker (batch-index ack is enabled by default on 4.0+ via `acknowledgmentAtBatchIndexLevelEnabled=true`, PIP-391; explicitly set here for clarity/for reproducing on 3.0.17): ```bash docker run -d -e PULSAR_PREFIX_acknowledgmentAtBatchIndexLevelEnabled=true apachepulsar/pulsar:4.0.11 \ sh -c 'bin/apply-config-from-env.py conf/standalone.conf && exec bin/pulsar standalone -nfw -nss' ``` Producer — sends 5 messages as a single batch: ```java PulsarClient client = PulsarClient.builder().serviceUrl(serviceUrl).build(); Producer<byte[]> producer = client.newProducer() .topic("persistent://public/default/negack-test") .batchingMaxPublishDelay(1, TimeUnit.SECONDS) .create(); List<CompletableFuture<?>> futures = new ArrayList<>(); for (int i = 0; i < 5; i++) { futures.add(producer.newMessage() .value(("message-" + i).getBytes(StandardCharsets.UTF_8)) .sendAsync()); } FutureUtil.waitForAll(futures).join(); ``` Consumer — `Shared` subscription, `ackTimeout(3s)`, `maxRedeliverCount(3)`. The app deliberately skips acknowledging batch indices 1 and 2 for the first 3 redelivery rounds (letting them expire via `ackTimeout` and get redelivered), then explicitly acks them once `redeliveryCount` reaches 3 (the last allowed round before DLQ routing): ```java Consumer<byte[]> consumer = client.newConsumer() .topic("persistent://public/default/negack-test") .subscriptionName("sub0") .subscriptionInitialPosition(SubscriptionInitialPosition.Latest) .subscriptionType(SubscriptionType.Shared) .enableBatchIndexAcknowledgment(true) .deadLetterPolicy(DeadLetterPolicy.builder().maxRedeliverCount(3).build()) .ackTimeout(3, TimeUnit.SECONDS) .subscribe(); while (true) { Message<byte[]> message = consumer.receive(); MessageId messageId = message.getMessageId(); if (messageId instanceof MessageIdAdv messageIdAdv && message.getRedeliveryCount() < 3 && (messageIdAdv.getBatchIndex() == 1 || messageIdAdv.getBatchIndex() == 2)) { System.out.printf("RedeliveryCount = %d, let %s timeout\n", message.getRedeliveryCount(), messageId); continue; } consumer.acknowledge(messageId); System.out.printf("RedeliveryCount = %d, acked %s\n", message.getRedeliveryCount(), messageId); } ``` Observed consumer-side log — batch indices 1 and 2 are explicitly acked at `RedeliveryCount = 3`, the final allowed round: ``` RedeliveryCount = 0, acked 6:0:-1:0 RedeliveryCount = 0, let 6:0:-1:1 timeout RedeliveryCount = 0, let 6:0:-1:2 timeout RedeliveryCount = 0, acked 6:0:-1:3 RedeliveryCount = 0, acked 6:0:-1:4 RedeliveryCount = 1, let 6:0:-1:1 timeout RedeliveryCount = 1, let 6:0:-1:2 timeout RedeliveryCount = 2, let 6:0:-1:1 timeout RedeliveryCount = 2, let 6:0:-1:2 timeout RedeliveryCount = 3, acked 6:0:-1:1 RedeliveryCount = 3, acked 6:0:-1:2 ``` A separate consumer subscribed directly to the DLQ topic (`negack-test-sub0-DLQ`) nevertheless receives those same two messages: ``` DLQ received: msgId=5:0:-1 value=message-1 DLQ received: msgId=5:1:-1 value=message-2 ``` `message-1` and `message-2` are exactly the batch indices 1 and 2 that the application acked at `RedeliveryCount = 3`. This was independently reproduced three times: Pulsar 3.0.17 (server+client), Pulsar 4.0.11 (server+client), and a broker built from current `master` source (commit `5951c36b7e3`) with a 4.0.11 client — same outcome in all three. ### Additional information ### Are you willing to submit a PR? Working on it, open a PR soon. - [x] I'm willing to submit a PR! -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
