apache / apache/pulsar

`receiver_queue_size` of 1 does not work; Shared consumers with receiver_queue_size=1 can "steal" messages that other consumers should receive

Open
#15,703 8 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

**Describe the bug**

When the Python client calls `subscribe`, setting `receiver_queue_size` to a value of 1 results in an effective receiver queue behavior of more than one; Shared-type consumers with `receiver_queue_size=1` can "steal" messages that other consumers could process.

**To Reproduce**
1. Create a persistent **non partitioned** topic.
2. Create a `Shared` subscription on the topic.
3. Publish 10 messages to the topic with sequential numeric payloads, e.g. `message-0`, `message-1` and so on.
4. Create a consumer on the topic with the below code, and ensure it prints `Press a key to acknowledge messages`.
5. In a second terminal, start another consumer on the topic with the below code, and ensure it prints `Press a key to acknowledge messages`.
6. Press "enter" in the first consumer's terminal.
7. Observe that the first consumer processes **fewer than 9** messages (in my tests it was usually 8, though I saw 6 occasionally; that might have been a defect on my side though).
8. Press "enter" in the second consumer's terminal.
9. Observe that the second consumer processes **more than 1** message.

Consumer code:
```python
from pulsar import Client, ConsumerType, Timeout
import os

TOPIC = 'THETOPIC'
SUBSCRIPTION = 'THESUBSCRIPTION'

def recv(sub):
while True:
try:
msg = sub.receive(100)
print("Got message", msg.data())
return msg
except Timeout:
pass

def main():
client = Client(service_url='pulsar://localhost:6650')
sub = client.subscribe(
topic=TOPIC,
subscription_name=SUBSCRIPTION,
consumer_type=ConsumerType.Shared,
receiver_queue_size=1,
consumer_name=f'testconsumer-{os.getpid()}'
)
msg = recv(sub)
mid = msg.message_id()
print("partition:", mid.partition(), "ledger:", mid.ledger_id(), "entry:", mid.entry_id(), "batch:", mid.batch_index())
input("Press a key to acknowledge messages")
while True:
sub.acknowledge(msg)
msg = recv(sub)

if __name__ == '__main__':
main()

```

**Expected behavior**

The first consumer should process exactly 9 messages, every time. The second consumer should process exactly 1 message, every time. The extra messages that go to consumer 2 should go to consumer 1; otherwise, consumer 2 can "falsely steal" those messages.

**Environment:**
Same environment as https://github.com/apache/pulsar-client-python/issues/190

Contributor guide

Open the contributing guide

Research direction

Start with the Python client's subscribe entry point and its receiver_queue_size handling. Reproduce the issue using the provided two-consumer Shared subscription script and verify how messages are distributed after acknowledgements. Done means receiver_queue_size=1 consistently leaves exactly 9 messages for the first consumer and 1 for the second.

Written by the indexing model from the issue text.

Assessment

Tech stack
python
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.