apache / apache/pulsar

Unexpected redelivery of a message

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

Description

**Describe the bug**
When a consumer is closing/closed and `redeliverUnacknowledgedMessages()` or `redeliverUnacknowledgedMessages(Set messageIds)` is called, then the code closes the client connection. This created unexpected behaviours for the other consumers. See the sample code below for instance

**To Reproduce**
Steps to reproduce the behavior:
```java
@Test
void simpletest() throws Exception {
PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(cluster.getAddress()).build();

ConsumerBuilder consumerBuilder = pulsarClient
.newConsumer(Schema.STRING)
.topic("test")
.subscriptionName("test")
.subscriptionType(SubscriptionType.Shared)
.negativeAckRedeliveryDelay(0, TimeUnit.SECONDS);

Consumer consumer1 = consumerBuilder.subscribe();
Consumer consumer2 = consumerBuilder.subscribe();

Producer producer =
pulsarClient.newProducer(Schema.STRING).topic("test").create();

producer.send("hello");
producer.send("world");

Message receive1 = consumer1.receive();
Message receive2 = consumer2.receive();

consumer1.negativeAcknowledge(receive1);
consumer1.close();

Thread.sleep(5000);

Message receive3 = consumer2.receive(5, TimeUnit.SECONDS);
Message receive4 = consumer2.receive(5, TimeUnit.SECONDS);

assertEquals("hello", receive3.getValue());
assertNull(receive4);
}
```

**Expected behavior**
This test should pass but it doesn't because the `pulsarClient` is reconnected, so `consumer2` is also reconnected and its unacked messages are redelivered. So `receive4` contains `world` instead of being null.

**Additional context**
The code that closes the connection is [here](https://github.com/apache/pulsar/blob/4147db88bc08eb3b2dab97a45af178fd9f4c7a6b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java#L1610). I'm not sure why we need to restart the connection here. If the consumer is closing/closed, the broker will deliver the unacked messages to another consumer anyway so why is it needed to reconnect completely the client ?

Contributor guide

Open the contributing guide

Research direction

Start with the reproducer in the issue and inspect the referenced redelivery logic in pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java around line 1610. Trace what happens when a consumer closes after negative acknowledgement and how reconnecting affects the other shared consumer. Done means the reproducer delivers the negatively acknowledged message to consumer2 without redelivering its already unacknowledged message.

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.