You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "Samreay (via GitHub)" <gi...@apache.org> on 2023/12/27 04:23:46 UTC
[I] Producer callback is unable to utilise python raise to propagate errors [pulsar-client-python]
Samreay opened a new issue, #184:
URL: https://github.com/apache/pulsar-client-python/issues/184
To reproduce:
1. Start a pulsar standalone instance, via
```
docker run -it -p 6650:6650 -p 8080:8080 --tmpfs /pulsar/data apachepulsar/pulsar:3.1.0 bin/pulsar standalone
```
2. Paste the following code into a file:
```python
import time
import pulsar
def callback(result: pulsar.Result, message_id: pulsar.MessageId):
if result == pulsar.Result.Timeout:
raise ValueError()
client = pulsar.Client("pulsar://localhost:6650")
producer = client.create_producer("topic-example", send_timeout_millis=1)
producer.send_async(b"hello", callback)
time.sleep(1)
```
3. Run the code. You should see something akin to:
```bash
...
2023-12-27 14:23:19.550 INFO [139722145527360] ProducerImpl:209 | [persistent://public/default/topic-example, ] Created producer on broker [127.0.0.1:56868 -> 127.0.0.1:6650]
terminate called after throwing an instance of 'pybind11::error_already_set'
what(): ValueError: <EMPTY MESSAGE>
At:
/home/sam/arenko/service-utils/tmp.py(8): callback
[1] 53944 IOT instruction /home/sam/arenko/service-utils/.venv/bin/python
```
Something about pybind and the C++ code means that errors cannot be raised in the callback function, which is obviously an issue for propagating errors up the python callstack.
Expected behaviour: The callback function acts as a standard python function with normal exception handling.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Producer callback is unable to utilise python raise to propagate errors [pulsar-client-python]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on issue #184:
URL: https://github.com/apache/pulsar-client-python/issues/184#issuecomment-1870909386
Did you mean the batch receive? Currently there is no good way in Pulsar to receive a batch as a whole. There is a `batchReceive` API in Java client and C++ client but both of them just receive message one by one and group them into an iterable batch.
It's because the raw format of a batch in Pulsar is not a trivial concatenation of message buffers. The `MessageMetadata` and `SingleMessageMetadata` protobuf structures are involved.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Producer callback is unable to utilise python raise to propagate errors [pulsar-client-python]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on issue #184:
URL: https://github.com/apache/pulsar-client-python/issues/184#issuecomment-1871191180
FYI, I'm going to support async-await style APIs, see https://github.com/apache/pulsar-client-python/issues/55#issuecomment-1871190583
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Producer callback is unable to utilise python raise to propagate errors [pulsar-client-python]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on issue #184:
URL: https://github.com/apache/pulsar-client-python/issues/184#issuecomment-1869966625
I got your point now. Unfortunately, the Pulsar Python client is not well integrated with Python coroutines. It's a callback-based solution rather than coroutine. As you can see, the `send_async` method **is not a method that can be await-ed**.
Therefore, you need a traditional way to handle it just like your workaround that queues the results.
BTW, I just found you have already checked the similar discussion: https://github.com/apache/pulsar-client-python/issues/55.
> Can this be documented at the very least then?
Sure.
> Additionally, doco on pulsar.Result types would be great
I agree. I opened another issue for it: https://github.com/apache/pulsar-client-python/issues/185
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Producer callback is unable to utilise python raise to propagate errors [pulsar-client-python]
Posted by "Samreay (via GitHub)" <gi...@apache.org>.
Samreay commented on issue #184:
URL: https://github.com/apache/pulsar-client-python/issues/184#issuecomment-1870664088
The workaround I have is more advanced than the one I posted above, and has proper concurrency when async sending multiple messages (unlike my example which sends one message at a time), and by having an awaitable task I can choose to wait for its completion via `await` or to just throw it into the event loop via `async.create_task(better_send(...))`.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Producer callback is unable to utilise python raise to propagate errors [pulsar-client-python]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower closed issue #184: Producer callback is unable to utilise python raise to propagate errors
URL: https://github.com/apache/pulsar-client-python/issues/184
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Producer callback is unable to utilise python raise to propagate errors [pulsar-client-python]
Posted by "Samreay (via GitHub)" <gi...@apache.org>.
Samreay commented on issue #184:
URL: https://github.com/apache/pulsar-client-python/issues/184#issuecomment-1870915794
So one of the goals for getting async working properly for my use case was that is seems the python producer can do batching but only with `send_async` and not `send`.
As an aside: it'd be great to know the batch send messages and bytes defaults and whether this should be set to 0 or -1 to be unlimited and _just_ batch off elapsed time, the doco for all those kwargs just says "Undocumented" https://pulsar.apache.org/api/python/3.3.x/pulsar.Client.html#create_producer
So every half an hour, we have a clearing action on a market, and we get a sudden and instantanous flood of messages from a third party. We turn these into pulsar messages, and want to batch send them out. Similarly, we want to batch read those messages in another service. Right now we dont use any of the batching functionality in the python consumer, because we need the interface between consumer and readers to be identical, and batching doesnt seem to be part of the reader implementation, just the consumer.
This means our pulsar wrapper just consumes messages one by one, with a little timeout after which all the messages its consumed get passed back to the function allocated to process the messages.
This all is just flavouring to answer that initial question of "Why not just use `send` instead of `send_async`?", and to boil the answer back its because `send` blocks and if I get 10k messages at once, sending them one by one isn't ideal, and sending them via `send_async` without ability to error handle is also not ideal, hence the current workaround. Hope that helps explain the use cases :)
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Producer callback is unable to utilise python raise to propagate errors [pulsar-client-python]
Posted by "merlimat (via GitHub)" <gi...@apache.org>.
merlimat commented on issue #184:
URL: https://github.com/apache/pulsar-client-python/issues/184#issuecomment-1870419432
@Samreay The question is also for which reason you are using `send_async()` instead of the plain `send`.
While the send is blocking, it won't be holding the GIL, so other Python threads can make progress.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Producer callback is unable to utilise python raise to propagate errors [pulsar-client-python]
Posted by "Samreay (via GitHub)" <gi...@apache.org>.
Samreay commented on issue #184:
URL: https://github.com/apache/pulsar-client-python/issues/184#issuecomment-1870896735
A bit more context:
Right now, if I have 100 messages to send and want actual error handling to exist, I either use my workaround or have to spend most of my time blocked using `send` because it seems the python client doesnt support batch sending by producers. I would be very happy to be wrong on this though! I have a batch consume on the same topic I want to produce in with another service, but using `send_sync` and each message being serial means instead of getting a single nice batch consumption on the end, I get numerous consumptions
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Producer callback is unable to utilise python raise to propagate errors [pulsar-client-python]
Posted by "Samreay (via GitHub)" <gi...@apache.org>.
Samreay commented on issue #184:
URL: https://github.com/apache/pulsar-client-python/issues/184#issuecomment-1869953268
Can this be documented at the very least then? Having a hidden "do this and your app will terminate" with no warning isn't ideal for an end user :) Additionally, doco on `pulsar.Result` types would be great, I struggled to handle this.
If you mean in terms of what should you see in the stack if the callback raises, I'd expect the `send_async` to be the root cause with the callback underneath it. If you mean in what execution context should a python async callback raise exceptions, that would be the event loop in general.
The goal, IMHO, is to get the async producer send as close to pythons standard library async as possible. Pythonically, this means you'd have something like:
```python
try:
result = await producer.send_async(b"hello")
except:
catch issues here
```
For a more complete example, this is my current workaround, which uses janus Queues (ie queues with sync and async interfaces) to convert the callback function into actual python async:
```python
import asyncio
from collections.abc import Awaitable
import janus
import pulsar
async def better_send(producer: pulsar.Producer, data: bytes) -> Awaitable[pulsar.MessageId]:
queue = janus.Queue()
def callback(result: pulsar.Result, message_id: pulsar.MessageId):
queue.sync_q.put((result, message_id))
producer.send_async(data, callback)
result, message_id = await queue.async_q.get()
if result == pulsar.Result.Timeout:
raise TimeoutError()
elif result != pulsar.Result.Ok:
raise ValueError(str(result))
return message_id
async def main():
client = pulsar.Client("pulsar://localhost:6650")
producer = client.create_producer("topic-example", send_timeout_millis=1)
await better_send(producer, b"hello")
if __name__ == "__main__":
asyncio.run(main())
```
Running this gives you a pythonic exception which doesn't crash the application, and has proper asyncio support.
```bash
Traceback (most recent call last):
File "/home/sam/arenko/service-utils/tmp.py", line 30, in <module>
asyncio.run(main())
File "/home/sam/.pyenv/versions/3.11.4/lib/python3.11/asyncio/runners.py", line 190, in run
return runner.run(main)
^^^^^^^^^^^^^^^^
File "/home/sam/.pyenv/versions/3.11.4/lib/python3.11/asyncio/runners.py", line 118, in run
return self._loop.run_until_complete(task)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/sam/.pyenv/versions/3.11.4/lib/python3.11/asyncio/base_events.py", line 653, in run_until_complete
return future.result()
^^^^^^^^^^^^^^^
File "/home/sam/arenko/service-utils/tmp.py", line 26, in main
await better_send(producer, b"hello")
File "/home/sam/arenko/service-utils/tmp.py", line 17, in better_send
raise TimeoutError()
TimeoutError
```
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Producer callback is unable to utilise python raise to propagate errors [pulsar-client-python]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on issue #184:
URL: https://github.com/apache/pulsar-client-python/issues/184#issuecomment-1869947644
You should not raise an exception in the callback because the callback is called in an internal thread.
> Expected behaviour: The callback function acts as a standard python function with normal exception handling.
If it's allowed to raise an exception in the callback, where will you expect to handle this exception?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Producer callback is unable to utilise python raise to propagate errors [pulsar-client-python]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on issue #184:
URL: https://github.com/apache/pulsar-client-python/issues/184#issuecomment-1869984264
> this would just cause even more issues with communication with the underlying C++ client
The root cause is that the C++ client (or the callback-based solution) does not provide a way for users to specify which thread to run the callback. What's worse, which thread calls the callback is a black box to users. Here is a similar issue I found: https://github.com/apache/pulsar-client-cpp/issues/368
The Python wrapper does not do some complicated work. Ideally, it should avoid user-provided callbacks being passed to the underlying C++ APIs directly, just like [what the Node.js wrapper did](https://github.com/apache/pulsar-client-node?tab=readme-ov-file#getting-started).
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Producer callback is unable to utilise python raise to propagate errors [pulsar-client-python]
Posted by "Samreay (via GitHub)" <gi...@apache.org>.
Samreay commented on issue #184:
URL: https://github.com/apache/pulsar-client-python/issues/184#issuecomment-1869969689
Yeah its a thorny issue here. There _is_ the option to add async support similar to how I've done it in the official client...
But ultimately, I worry this would just cause even more issues with communication with the underlying C++ client, and this probably brings the discussion back to "Should the python client be rewritten (but a vague 'someone') in python to use standard language features?"
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Producer callback is unable to utilise python raise to propagate errors [pulsar-client-python]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on issue #184:
URL: https://github.com/apache/pulsar-client-python/issues/184#issuecomment-1871023180
> it'd be great to know the batch send messages and bytes defaults and whether this should be set to 0 or -1 to be unlimited and just batch off elapsed time
Yeah these documents are missed, we should add them. For now, you can refer the [C++ client documents here](https://pulsar.apache.org/api/cpp/3.4.x/classpulsar_1_1_producer_configuration.html) for all `setBatchXXX` methods. In short, 0 means no limit. (I'm not sure about the behavior of setting with -1 because of the signed and unsigned issue)
Here is a related issue: https://github.com/apache/pulsar-client-python/issues/187
----
BTW, backing to the issue title. sing a callback just forces users to pass the errors in another thread to process, which is generally implemented via a concurrent queue.
Regarding to your concern before:
> But ultimately, I worry this would just cause even more issues with communication with the underlying C++ client
I think it's okay to add such APIs just like your workaround. The underlying C++ client calls the send callback in Boost.Asio's event loop, which just works like `asyncio`'s event loop.
My brainstorm:
```python
producer = client.create_producer(topic, max_pending_sends=5)
coros = list()
for i in range(10):
try:
coros.append(producer.send_async("msg"))
except pulsar.QueueIsFull as e:
break # len(coros) is 5 here
try:
msg_ids = await asyncio.gather(*coros) # or wait here
except pulsar.PulsarException as e:
# ...
except Exception as e:
# ...
```
The internal implementation just uses a queue that supports synchronous put and asynchronous get like `janus.Queue` you use.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Producer callback is unable to utilise python raise to propagate errors [pulsar-client-python]
Posted by "merlimat (via GitHub)" <gi...@apache.org>.
merlimat commented on issue #184:
URL: https://github.com/apache/pulsar-client-python/issues/184#issuecomment-1872156345
@Samreay
The way to achieve what you're describing with the current client is like:
```python
producer = client.create_producer(topic, block_if_queue_full=True)
for i in range(100):
producer.send_async("msg")
producer.flush()
```
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Producer callback is unable to utilise python raise to propagate errors [pulsar-client-python]
Posted by "Samreay (via GitHub)" <gi...@apache.org>.
Samreay commented on issue #184:
URL: https://github.com/apache/pulsar-client-python/issues/184#issuecomment-1873543366
Thanks @merlimat, I didn't realise flush would raise errors like that - very handy and I'll add this into my code to try and increase its robustness :)
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org