Getting this package work with asyncio #185
Comments
|
Hi, Stay tuned! related issue: |
|
Are there any updates on this? Is anyone working on it? |
|
+1 |
|
Not yet, community contributions are very much welcome. |
|
+1 |
|
In the meantime, https://github.com/aio-libs/aiokafka is a python / kafka library that supports |
|
Hi, just checking, is it yet compatiable with asyncio |
|
Hi, I started looking at this a bit. asyncio has the ability to watch file descriptors for read availability (https://docs.python.org/3/library/asyncio-eventloop.html#watch-file-descriptors) librdkafka can write to a file descriptor when a queue is readable with Would it be possible to expose Ideally you should be able to pass a file descriptor and a payload. When the file descriptor becomes readable, call There are still a handful of synchronous operations (like asyncio support could probably be integrated more directly in this library, but the above might be a good starting point to allow for an asycio wrapper. |
|
Since go-lang wrapper can benefit |
|
Any updates here? |
|
+1 |
1 similar comment
|
+1 |
|
+1 |
|
Leaving this here for reference until this feature gets implemented. It's possible to process the messages with Asyncio it seems. https://stackoverflow.com/a/55498503 I have not tried this out myself yet but might be of help. |
|
I guess some people might just want it working with asyncio, not working with maximum efficiency. So here is an example that works now: import asyncio
async def run_consumer(consumer):
while True:
msg = consumer.poll(timeout=0)
if msg is None:
await asyncio.sleep(delay=0)
continue
# handle msg
consumer.commit(msg, asynchronous=True)
await run_consumer(consumer) |
|
The main issue is the support of the Schema Registry, this one does not exist with aiokafka. |
Yes, but keep in mind that this API is not very portable, e.g. it does not work with Windows ProactorEventLoop. |
|
keep an eye on the confluent blog over the next couple of weeks for an in depth blog post on this client and asyncio. it walks through the producer only, though. @h12w - your solution will busy loop and will be very inefficient. to be efficient, you need to do blocking consumer calls, and to do that you'll need another thread. |
|
@mhowlett any update on the detailed blog.. couldn't search for it |
|
Has anyone been able to successfully connect to My adapted version of the code is: import asyncio
import os
import logging
from aiokafka import AIOKafkaProducer, AIOKafkaConsumer
from aiokafka.helpers import create_ssl_context
from kafka.common import TopicPartition
from aiokafka.errors import KafkaError
from aiokafka import AIOKafkaClient
import ccloud_lib
conf = ccloud_lib.read_ccloud_config('kafka_config.conf')
ssl_context = create_ssl_context(cafile='cacert.pem')
log_level = logging.DEBUG
log_format = '[%(asctime)s] %(levelname)s [%(name)s]: %(message)s'
logging.basicConfig(level=logging.DEBUG, format=log_format)
async def produce_and_consume(loop):
# Produce
producer = AIOKafkaProducer(
bootstrap_servers=conf['bootstrap.servers'],
loop = loop,
security_protocol=conf['security.protocol'],
sasl_mechanism=conf['sasl.mechanism'],
ssl_context=ssl_context,
sasl_plain_username=conf['sasl.username'],
sasl_plain_password=conf['sasl.password'],
api_version='0.10'
)
try:
start_future = await producer.start()
response = await start_future # wait until message is produced
except KafkaError as err:
print("some kafka error on produce: {}".format(err))
try:
msg = await producer.send_and_wait(
'my_topic', b"Super Message", partition=0)
finally:
await producer.stop()
consumer = AIOKafkaConsumer(
bootstrap_servers=conf['bootstrap.servers'],
loop=loop,
ssl_context=ssl_context,
security_protocol=conf['security.protocol'],
sasl_mechanism=conf['sasl.mechanism'],
sasl_plain_password=conf['sasl.password'],
sasl_plain_username=conf['sasl.username']
)
try:
start_future = await consumer.start()
response = await start_future # wait until message is produced
except KafkaError as err:
print("some kafka error on produce: {}".format(err))
try:
consumer.seek(TopicPartition('my_topic', 0), msg.offset)
fetch_msg = await consumer.getone()
finally:
await consumer.stop()
print("Success", msg, fetch_msg)
if __name__ == "__main__":
loop = asyncio.get_event_loop()
task = loop.create_task(produce_and_consume(loop))
try:
loop.run_until_complete(task)
finally:
loop.run_until_complete(asyncio.sleep(0, loop=loop))
task.cancel()
try:
loop.run_until_complete(task)
except asyncio.CancelledError:
passMy obfuscated configuration Is there something about my configuration which is incorrect or missing? |
|
@galen211 you should post the issue on https://github.com/aio-libs/aiokafka/issues or email confluent support... this repo is unrelated to either that lib or that cloud. |
|
As @jeffwidman pointed out you will need to take this up with aiokafka. I'm closing this issue with @mhowlett's blog being the resolution. |
and others
Hey guys,
great job on this package!
I wanted to know if there is a way to use asyncio with this package, since I saw some code that using "async await" thingy
If does, there is any example?
thanks,
Amit
The text was updated successfully, but these errors were encountered: