apache / apache/pulsar

acknowledgmentAtBatchIndexLevelEnabled will still receive duplicate messages when unload topic

Open
#14,982 11 comments 0 reactions 0 assignees View on GitHub
lifecycle/stale Stale type/bug
Dominant language
Java
Stars
15.3k
Forks
3.8k
Avg merge
1d 14h
Merged PRs (30d)
160

Description

**Describe the bug**
Pulsar v2.8.3
acknowledgmentAtBatchIndexLevelEnabled will still receive duplicate messages

**To Reproduce**
Steps to reproduce the behavior:
1. Set acknowledgmentAtBatchIndexLevelEnabled=true in broker.conf
2. Create consumer:
```
public static void main(String[] args) throws Exception{
PulsarClient client = PulsarClient
.builder()
.serviceUrl("pulsar://127.0.0.1:6650")
.build();

ConsumerBuilder consumerBuilder = client.newConsumer(Schema.JSON(NlMessage.class)).topic("batchtest")
.subscriptionName("batchTest")
.subscriptionType(SubscriptionType.Shared)
.enableRetry(true)
.enableBatchIndexAcknowledgment(true)
.receiverQueueSize(1000)
.negativeAckRedeliveryDelay(20, TimeUnit.SECONDS)
.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest);
// DLQ
DeadLetterPolicy deadLetterPolicy = DeadLetterPolicy.builder()
.deadLetterTopic("persistent://public/default/batchtest-dlq")
.retryLetterTopic("persistent://public/default/batchtest-retry")
.maxRedeliverCount(3)
.build();
consumerBuilder.deadLetterPolicy(deadLetterPolicy);
consumerBuilder.negativeAckRedeliveryDelay(10, TimeUnit.MICROSECONDS);
consumerBuilder.ackTimeout(0, TimeUnit.SECONDS);

Consumer consumer = consumerBuilder.subscribe();
long i = 0;
while (true) {
Message message = consumer.receive();
System.out.println("message:" + message.getMessageId());
if (i <= 99) {
consumer.acknowledge(message);
System.out.println("acked message:" + i);
} else {
// hold the msg
}
i++;
}
```
3. Create producer and send 100 msgs:
```
public static void main(String[] args) throws PulsarClientException {
PulsarClient pulsarClient = PulsarClient.builder().serviceUrl("pulsar://127.0.0.1:6650").build();
Producer producer = pulsarClient.newProducer(Schema.JSON(NlMessage.class))
.topic("persistent://public/default/batchtest")
.blockIfQueueFull(true)
.sendTimeout(0, TimeUnit.SECONDS)
.create();
long i = 1;
while(i <= 100L){
producer.sendAsync(new NlMessage());
i++;
}
}
```
4. 100 msgs in 1 backlog
![image](https://user-images.githubusercontent.com/54351417/161229419-14ebd41e-35fa-4fd4-abf2-2ab62097c5a3.png)
5. After 2 minutes, execute unload cmd:
`bin/pulsar-admin topics unload persistent://public/default/batchtest
`
6. See consumer log, there will also be several or more messages that have been ack pushed.

**Expected behavior**
Only push the message without ack.

**Screenshots**
If applicable, add screenshots to help explain your problem.

**Desktop (please complete the following information):**
- OS: [e.g. iOS]

**Additional context**
Add any other context about the problem here.

Contributor guide

Open the contributing guide

Research direction

Start by reproducing the behavior with acknowledgmentAtBatchIndexLevelEnabled=true, the shown batch-acknowledgment consumer, and 100 messages. Run the unload command for the topic and inspect the consumer log. Done means unloading does not redeliver messages that were already acknowledged; only unacknowledged messages are pushed again.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
distributed-systems
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.