[Bug] Consumer receives duplicate messages even after acknowledging them.
- 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
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