Implement async confluent-kafka producer #775
Merged
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
matyaskuti
force-pushed
the
matyaskuti/confluent_kafka_asyncio
branch
from
a5e4bbe to
7e6d22f
Compare
matyaskuti
force-pushed
the
matyaskuti/confluent_kafka_sync_consumer
branch
from
4740bab to
0112a8c
Compare
matyaskuti
force-pushed
the
matyaskuti/confluent_kafka_asyncio
branch
2 times, most recently
from
56642a0 to
cdd9907
Compare
matyaskuti
force-pushed
the
matyaskuti/confluent_kafka_sync_consumer
branch
from
0112a8c to
981fa69
Compare
matyaskuti
force-pushed
the
matyaskuti/confluent_kafka_asyncio
branch
from
cdd9907 to
3a2fa65
Compare
matyaskuti
force-pushed
the
matyaskuti/confluent_kafka_sync_consumer
branch
from
981fa69 to
0039127
Compare
matyaskuti
force-pushed
the
matyaskuti/confluent_kafka_asyncio
branch
2 times, most recently
from
f112c8f to
334035e
Compare
matyaskuti
force-pushed
the
matyaskuti/confluent_kafka_sync_consumer
branch
from
0039127 to
f86d68d
Compare
matyaskuti
force-pushed
the
matyaskuti/confluent_kafka_asyncio
branch
from
334035e to
f5e9ef3
Compare
matyaskuti
changed the title
matyaskuti
force-pushed
the
matyaskuti/confluent_kafka_asyncio
branch
5 times, most recently
from
096fda4 to
8eeade1
Compare
matyaskuti
force-pushed
the
matyaskuti/confluent_kafka_sync_consumer
branch
13 times, most recently
from
8e0f0d4 to
e91aafa
Compare
eliax1996
reviewed
Jan 18, 2024
Just to make sure this doesn't become a future blocker, it could perhaps be worth experimenting with handling SSL external to confluent-kafka, and making it communicate through an already setup TLS connection. |
aiven-anton
reviewed
Jan 18, 2024
matyaskuti
force-pushed
the
matyaskuti/confluent_kafka_asyncio
branch
4 times, most recently
from
3a3f265 to
5787118
Compare
aiven-anton
reviewed
Jan 18, 2024
matyaskuti
force-pushed
the
matyaskuti/confluent_kafka_asyncio
branch
11 times, most recently
from
89db3f9 to
19d8b16
Compare
aiven-anton
previously approved these changes
Jan 22, 2024
added 2 commits
January 22, 2024 11:01
THe async `AIOKafkaProducer` is implemented as a wrapper around `KafkaProducer` with an async `send` method and a poll-thread to continuously send messages in the background, making the result of `send` an awaitable `asyncio.Future`. The example of confluent-kafka has been followed: https://github.com/confluentinc/confluent-kafka-python/blob/master/examples/asyncio_example.py
REST webapp improvement: Using `on_cleanup` is not the correct hook to use here, as this would run _after_ the event loop has closed, making it unsuitable for cancelling background tasks for example. `on_shutdown` is triggered before the REST app shuts down, thus it's able to clean up eg. Kafka clients, background tasks, etc. properly. Before this change, the symptom of the bug is most prevalent in the Karapace REST proxy and its "idle proxy janitor" background task. Stopping the application when the janitor task is not running is straightforward, however when any `UserRestProxy` is present (ie. some requests have already been handled) and the task is running, stopping the REST proxy hangs or needs multiple signals to shut down. With the new `AIOKafkaProducer` implementation (which runs a poll-thread in the background) this results in an application that is unable to gracefully shutdown, only SIGKILL works. Using the `on_shutdown` hook fixes this issue, as we still have an event loop available to be able to cancel background tasks, etc.
matyaskuti
force-pushed
the
matyaskuti/confluent_kafka_asyncio
branch
from
19d8b16 to
a51027d
Compare
aiven-anton
approved these changes
Jan 22, 2024
eliax1996
approved these changes
Jan 22, 2024
jjaakola-aiven
approved these changes
Jan 22, 2024
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
About this change - What it does
KafkaProducerwith an asyncsendmethod and a poll-thread to send messages continuously in the backgroundon_cleanuptoon_shutdown- this was an existing bug preventing any kind of closing of resources (Kafka clients, consumer manager, etc.) to actually run, further explanation in commit messageWhy this way