apache / apache/pulsar

[Bug] Consumer receives duplicate messages even after acknowledging them.

Open
#23,436 2 comments 0 reactions 0 assignees View on GitHub
triage/lhotari/important type/bug
Dominant language
Java
Stars
15.3k
Forks
3.8k
Avg merge
1d 14h
Merged PRs (30d)
160

Description

### Search before asking

- [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 don't get bug fixes. I will attempt to reproduce the issue on a supported version of Pulsar client and Pulsar broker.

### Version

All version

### Minimal reproduce step

1. Start standalone pulsar
2. produce some message(enable batch)
``` java
// Create a client
PulsarClient client = PulsarClient.builder()
.serviceUrl("pulsar://localhost:6650")
.build();

// Create a producer
ProducerBuilder producerBuilder = client.newProducer(Schema.BYTES)
.topic("persistent://public/default/my-topic")
.sendTimeout(30, java.util.concurrent.TimeUnit.SECONDS)
.enableBatching(true)
.batchingMaxMessages(60)
.blockIfQueueFull(true);

Producer producer = producerBuilder.create();

// Send messages
for (int i = 0; i < 60; i++) {
String msg = "my-message-" + i;
producer.send(msg.getBytes());
System.out.println("Sent message: " + i);
}

producer.flush();
producer.close();
client.close();
```
4. Consume these message

``` java
public static void main(String[] args) throws Exception {
PulsarClient client = PulsarClient.builder()
.serviceUrl("pulsar://localhost:6650")
.build();

Consumer consumer = client.newConsumer()
.topic("persistent://public/default/my-topic")
.subscriptionName("test-sub")
.subscriptionType(SubscriptionType.Shared)
.subscribe();

// Set to store processed messageIds
Set processedMessageIds = new HashSet<>();

while (true) {
System.out.println("Start call batch receive");
Message message = consumer.receive();
MessageId messageId = message.getMessageId();
String messageData = new String(message.getData());
System.out.printf("Received message: %s: %s%n", messageId.toString(), messageData);

if ("my-message-59".equals(messageData)) {
consumer.negativeAcknowledge(message);
System.out.printf("Nack acknowledge %s messages%n", messageId.toString());
} else {
consumer.acknowledge(message);
System.out.printf("Acknowledge %s messages%n", messageId.toString());
}

if (processedMessageIds.contains(messageId)) {
System.out.printf("Received duplicate messageId: %s%n", messageId.toString());
} else {
processedMessageIds.add(messageId);
}

Thread.sleep(10);
}
}
```

### What did you expect to see?

Don't print messages 1 to 58, only print message 59.

### What did you see instead?

All messages are being received repeatedly.

### Anything else?

1. I know enabling batchIndexAck can resolve this issue.
2. However, it's confusing for users because they acknowledge some messages, yet they still receive duplicates.
3. We have deduplication logic: https://github.com/apache/pulsar/pull/21116, but it's not utilized if the client doesn't enable `batchIndexAck`.
4. Even if the client enables batchIndexAck, if the broker doesn't, duplicate messages will still be received. This is because when the client sends the ack to the broker, the broker ignores it, causing the client's bitset to be cleared, resulting in duplicate messages.

### Are you willing to submit a PR?

- [ ] I'm willing to submit a PR!

Contributor guide

Open the contributing guide

Research direction

Start with the supplied Java standalone producer and consumer reproduction, then trace acknowledgement handling with batchIndexAck enabled and disabled on both client and broker. Done means acknowledged messages 1–58 are not redelivered while negatively acknowledged message 59 is, with regression coverage for both settings.

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
38/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.