scheung38 opened a new issue #7636:
URL: https://github.com/apache/pulsar/issues/7636
Started pulsar (2.6.0)
PyCharm
(venv) python --version
Python 3.8.4
$ pulsar standalone
sample-producer.py:
```
import pulsar
client = pulsar.Client('pulsar://localhost:6650')
producer = client.create_producer('my-topic')
for i in range(10):
producer.send(('Hello-Sebastian-%d' % i).encode('utf-8'))
client.close()
```
$ python producer.py
2020-07-22 22:09:47.363 INFO [0x11bc99dc0] ConnectionPool:85 | Created
connection for pulsar://localhost:6650
2020-07-22 22:09:47.365 INFO [0x700008b1b000] ClientConnection:335 |
[127.0.0.1:63067 -> 127.0.0.1:6650] Connected to broker
2020-07-22 22:09:47.369 INFO [0x700008b1b000] HandlerBase:53 |
[persistent://public/default/my-topic, ] Getting connection from pool
2020-07-22 22:09:47.372 INFO [0x700008b1b000] ProducerImpl:165 |
[persistent://public/default/my-topic, ] Created producer on broker
[127.0.0.1:63067 -> 127.0.0.1:6650]
2020-07-22 22:09:47.469 INFO [0x11bc99dc0] ClientImpl:481 | Closing Pulsar
client
2020-07-22 22:09:47.469 INFO [0x11bc99dc0] ProducerImpl:502 |
[persistent://public/default/my-topic, standalone-4-0] Closing producer for
topic persistent://public/default/my-topic
2020-07-22 22:09:47.473 INFO [0x700008b1b000] ProducerImpl:545 |
[persistent://public/default/my-topic, standalone-4-0] Closed producer
2020-07-22 22:09:47.473 INFO [0x700008b1b000] ClientConnection:1372 |
[127.0.0.1:63067 -> 127.0.0.1:6650] Connection closed
2020-07-22 22:09:47.476 INFO [0x11bc99dc0] ProducerImpl:483 | Producer -
[persistent://public/default/my-topic, standalone-4-0] , [batching = off]
2020-07-22 22:09:47.476 INFO [0x11bc99dc0] ClientConnection:235 |
[127.0.0.1:63067 -> 127.0.0.1:6650] Destroyed connection
sample-consumer.py:
```
import pulsar
client = pulsar.Client('pulsar://localhost:6650')
consumer = client.subscribe('my-topic', 'my-subscription')
while True:
msg = consumer.receive()
try:
print("Received message '{}' id='{}'".format(msg.data(),
msg.message_id()))
# Acknowledge successful processing of the message
consumer.acknowledge(msg)
except:
consumer.negative_acknowledge(msg)
client.close()
```
$ python consumer.py
2020-07-22 22:11:22.367 INFO [0x1164a1dc0] Client:88 | Subscribing on Topic
:my-topic
2020-07-22 22:11:22.367 INFO [0x1164a1dc0] ConnectionPool:85 | Created
connection for pulsar://localhost:6650
2020-07-22 22:11:22.369 INFO [0x700001922000] ClientConnection:335 |
[127.0.0.1:63138 -> 127.0.0.1:6650] Connected to broker
2020-07-22 22:11:22.372 INFO [0x700001922000] AckGroupingTrackerEnabled:53
| ACK grouping is enabled, grouping time 100ms, grouping max size 1000
2020-07-22 22:11:22.372 INFO [0x700001922000] HandlerBase:53 |
[persistent://public/default/my-topic, my-subscription, 0] Getting connection
from pool
2020-07-22 22:11:22.374 INFO [0x700001922000] ConsumerImpl:199 |
[persistent://public/default/my-topic, my-subscription, 0] Created consumer on
broker [127.0.0.1:63138 -> 127.0.0.1:6650]
Received message 'b'Hello-0'' id='(294,10,-1,-1)'
Received message 'b'Hello-1'' id='(294,11,-1,-1)'
Received message 'b'Hello-2'' id='(294,12,-1,-1)'
Received message 'b'Hello-3'' id='(294,13,-1,-1)'
Received message 'b'Hello-4'' id='(294,14,-1,-1)'
Received message 'b'Hello-5'' id='(294,15,-1,-1)'
Received message 'b'Hello-6'' id='(294,16,-1,-1)'
Received message 'b'Hello-7'' id='(294,17,-1,-1)'
Received message 'b'Hello-8'' id='(294,18,-1,-1)'
Received message 'b'Hello-9'' id='(294,19,-1,-1)'
First time ran ok but making new messages and reran producer and consumer
second time:
sample-consumer.py
2020-07-22 22:17:04.911 INFO [0x1156addc0] Client:88 | Subscribing on Topic
:my-topic
2020-07-22 22:17:04.912 INFO [0x1156addc0] ConnectionPool:85 | Created
connection for pulsar://localhost:6650
2020-07-22 22:17:04.914 INFO [0x700008d88000] ClientConnection:335 |
[127.0.0.1:63330 -> 127.0.0.1:6650] Connected to broker
Traceback (most recent call last):
2020-07-22 22:17:04.917 INFO [0x700008d88000] AckGroupingTrackerEnabled:53
| ACK grouping is enabled, grouping time 100ms, grouping max size 1000
File
"/Users/mincheung/PycharmProjects/pulsar-sample/2_sample-consumer.py", line 5,
in <module>
2020-07-22 22:17:04.917 INFO [0x700008d88000] HandlerBase:53 |
[persistent://public/default/my-topic, my-subscription, 0] Getting connection
from pool
consumer = client.subscribe('my-topic', 'my-subscription')
2020-07-22 22:17:04.918 WARN [0x700008d88000] ClientConnection:939 |
[127.0.0.1:63330 -> 127.0.0.1:6650] Received error response from server:
ConsumerBusy -- req_id: 0
File
"/Users/mincheung/PycharmProjects/pulsar-sample/venv/lib/python3.8/site-packages/pulsar/__init__.py",
line 655, in subscribe
2020-07-22 22:17:04.918 ERROR [0x700008d88000] ConsumerImpl:242 |
[persistent://public/default/my-topic, my-subscription, 0] Failed to create
consumer: ConsumerBusy
c._consumer = self._client.subscribe(topic, subscription_name, conf)
Exception: Pulsar error: ConsumerBusy
2020-07-22 22:17:04.918 WARN [0x700008d88000] AckGroupingTrackerEnabled:99
| Connection is not ready, grouping ACK failed.
2020-07-22 22:17:04.922 INFO [0x1156addc0] ClientConnection:1372 |
[127.0.0.1:63330 -> 127.0.0.1:6650] Connection closed
2020-07-22 22:17:04.922 INFO [0x1156addc0] ClientConnection:235 |
[127.0.0.1:63330 -> 127.0.0.1:6650] Destroyed connection
Process finished with exit code 1
Unless I close Terminal and run consumer in new Terminal to be successful?
**Desktop (please complete the following information):**
- OS: Mac OSX Catalina 10.15.5
**Additional context**
Add any other context about the problem here.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
