`receiver_queue_size` of 1 does not work; Shared consumers with receiver_queue_size=1 can "steal" messages that other consumers should receive
- 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
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