Add Consumer.io_event_enable #1448
Conversation
|
|
stephan-hof
force-pushed
the
features/io_event_enable
branch
2 times, most recently
from
8fbac8d to
9a73aa8
Compare
Having this as an example is good 👍 |
| @@ -0,0 +1,164 @@ | |||
| import argparse | |||
There was a problem hiding this comment.
Add confluent copyright header (see other files), update year to 2022
|
|
||
| assert sys.platform == 'linux', "This example is linux only, cause of eventfd" | ||
|
|
||
|
|
There was a problem hiding this comment.
It'd be good with a bit of text here (or after the license) which explains what this example is for, and why.
examples/linux_asyncio_consumer.py
Outdated
| def __init__(self, config, logger): | ||
| self.consumer = Consumer(config, logger=logger) | ||
|
|
||
| # Sorry Windows/MacOX try something with socketpair or pipe. |
There was a problem hiding this comment.
| # Sorry Windows/MacOX try something with socketpair or pipe. | |
| # Sorry Windows/MacOX try something with socketpair or pipe. |
| # Sorry Windows/MacOX try something with socketpair or pipe. | |
| # FIXME: Windows/MacOX try something with socketpair or pipe. |
| self.consumer.assign(*args, **kwargs) | ||
|
|
||
| async def poll(self, timeout=0): | ||
| if timeout > 0: |
There was a problem hiding this comment.
This warrants a doc string, since with async it differs from the usual poll
examples/linux_asyncio_consumer.py
Outdated
| config = { | ||
| 'bootstrap.servers': arguments.bootstrap_servers, | ||
| 'group.id': arguments.group_id, | ||
| 'session.timeout.ms': 6000, |
There was a problem hiding this comment.
This is not a recommended value, it will give poor robustness, so please remove.
src/confluent_kafka/src/Consumer.c
Outdated
| Py_ssize_t data_size = 0; | ||
| static char *kws[] = {"fd", "payload", NULL}; | ||
|
|
||
| if (!PyArg_ParseTupleAndKeywords(args, kwargs, "iy#", kws, |
There was a problem hiding this comment.
see other functions that check self->rk first.
src/confluent_kafka/src/Consumer.c
Outdated
| METH_VARARGS | METH_KEYWORDS, | ||
| ".. py:function:: io_event_enable(fd, payload)\n" | ||
| "\n" | ||
| "This method basically calls ``rd_kafka_queue_io_event_enable``\n" |
There was a problem hiding this comment.
We try to abstract the underlying C library, so this doesn't tell the user much, so skip this line altogether.
src/confluent_kafka/src/Consumer.c
Outdated
| "Enable IO event triggering.\n" | ||
| "To ease integration with IO based polling loops this API\n" | ||
| "allows an application to create a separate file-descriptor\n" | ||
| "that librdkafka will write ``payload`` to whenever a new\n" |
There was a problem hiding this comment.
| "that librdkafka will write ``payload`` to whenever a new\n" | |
| "that the consumer will write ``payload`` to whenever a new\n" |
src/confluent_kafka/src/Consumer.c
Outdated
| "allows an application to create a separate file-descriptor\n" | ||
| "that librdkafka will write ``payload`` to whenever a new\n" | ||
| "element is enqueued on a previously empty queue.\n" | ||
| " :param int fd: The filedescrptor librdkafka writes to.\n" |
There was a problem hiding this comment.
| " :param int fd: The filedescrptor librdkafka writes to.\n" | |
| " :param int fd: The filedescrptor the consumer writes to.\n" |
src/confluent_kafka/src/Consumer.c
Outdated
| "that librdkafka will write ``payload`` to whenever a new\n" | ||
| "element is enqueued on a previously empty queue.\n" | ||
| " :param int fd: The filedescrptor librdkafka writes to.\n" | ||
| " :param bytes payload: The payload librdkakfa writes to fd.\n" |
There was a problem hiding this comment.
| " :param bytes payload: The payload librdkakfa writes to fd.\n" | |
| " :param bytes payload: The payload the consumer writes to fd.\n" |
|
Many thanks for looking at this pull request. I made a commit trying to address the changes requested by the review. I'm wondering now what is the next step on my side. Should I mark the conversations as 'resolved'? |
|
We're currently in release code freeze, but will revisit this PR in January when the upcoming release is out. |
|
Any update on this PR? I'd like to use Confluent Kafka's asynchronous methods in https://github.com/faust-streaming/faust |
stephan-hof
force-pushed
the
features/io_event_enable
branch
from
97b3a87 to
cd0f5b2
Compare
|
Thanks for the ping. In the hope to speed-up things I rebased the branch on latest master. |
|
Bumping for posterity. Is the code freeze over @edenhill? We would really like this functionality to be implemented to use in Faust and many other asynchronous Kafka applications! |
stephan-hof
force-pushed
the
features/io_event_enable
branch
from
cd0f5b2 to
69b79ad
Compare
|
It would be amazing to have :) @edenhill |
|
It appears to me that @edenhill is not very active in the last months (hope you are well). @pranavrth I see that you keep this project alive. So is this pull request something you cold have a look at please? |
|
Is there any other maintainer who can take a look at this 🙏🏻 ? This is a huge win for async kafka |
|
bump any chance to have asyncio in confluent-kafka? |
Hello,
this pull requests adds a method called
io_event_enableto theConsumer. It simply wrapsrd_kafka_queue_io_event_enablefrom librdkafka.My goal is to make integration with async frameworks easier. Using it allows the async framework to be immediately notified when a new message arrived in the queue. Avoiding,
Consumer.poll(timeout=0)I'm not entirely sure about the
linux_asyncio_consumer.py.Should it be part of the confluent-kafka repository or not? My thinking was to show how
io_event_enablecould be used.Update:
Now that I checked the issue tracker somebody already came up with the idea: #185 (comment) so consider this pull-request as an implementation of it.