apache / apache/pulsar

[Bug] Delayed message delivery receive duplicate messages when unload topic

Open
#20,099 16 comments 0 reactions 0 assignees View on GitHub
Stale 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.

### Version

2.7.5 / 2.8.4 /master

### Minimal reproduce step

1. Run the code for 20 minutes

```java
public static void main(String[] args) {
String url = "pulsar://10.98.242.210:6650";
String t = "persistent://components/test/test_deliver";
new Thread(() -> {
try {
PulsarClient client = PulsarClient.builder()
.serviceUrl(url)
.ioThreads(1)
.listenerThreads(2)
.build();
Producer producerA = client.newProducer(Schema.STRING)
.topic(t)
.producerName("test")
.create();
Random random = new Random();
while (true) {
String message = String.valueOf(System.currentTimeMillis());
producerA.newMessage(Schema.STRING).value(message).deliverAfter(random.nextInt(30), TimeUnit.MINUTES).send();
Thread.sleep(10L);
}
} catch (Exception e) {
e.printStackTrace();
}
}).start();
Set stringSet = new ConcurrentHashSet<>();
try {
PulsarClient client = PulsarClient.builder()
.serviceUrl(url)
.ioThreads(1)
.listenerThreads(2)
.build();
Consumer consumer = client.newConsumer(Schema.STRING)
.topic(t)
.subscriptionName("frozen")
.subscriptionType(SubscriptionType.Shared)
.subscribe();
while (true) {
Message message = consumer.receive();
if (stringSet.contains(message.getValue())) {
log.info("duplicate messages thread={} message={}", Thread.currentThread().getName(), message.getValue());
} else {
stringSet.add(message.getValue());
}
consumer.acknowledge(message);
}
} catch (Exception e) {
e.printStackTrace();
}
}
```

2. After 20 minutes, execute unload cmd:

```
bin/pulsar-admin topics unload persistent://components/test/test_deliver
```

3. See consumer log, duplicate messages

### What did you expect to see?

no duplicate messages

### What did you see instead?

duplicate messages

### Anything else?

_No response_

### Are you willing to submit a PR?

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

Contributor guide

Open the contributing guide

Research direction

Start by running the Java reproducer with delayed messages, then execute the documented pulsar-admin topics unload command and compare consumer deliveries before and after unloading. Trace the delayed-message and topic-unload handling involved in this scenario; done means unloading the topic no longer causes already scheduled messages to be delivered more than once.

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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.