aiokafka
aiokafka copied to clipboard
aio-libs
asyncio client for kafka
The current implementation of the aiokafka library within the Confluent platform only supports client versions under 3.0 for Python. There is an error reported in [aiokafka/util.py](https://github.com/aio-libs/aiokafka/blob/199537bb02177ff935ccdefa2b44615214024a0f/aiokafka/util.py#L59) that restricts compatibility. Additionally,...
### Changes Fixes # ### Checklist - [ ] I think the code is well written - [ ] Unit tests for the changes exist - [ ] Documentation reflects...
### Changes Fixes # ### Checklist - [ ] I think the code is well written - [ ] Unit tests for the changes exist - [ ] Documentation reflects...
**Describe the bug** Whenever AIOKafkaConsumer (maybe Producer too) loses connection to Kafka, 17 identical error logs are outputted. ``` Failed fetch messages from 1: NodeNotReadyError: Attempt to send a request...
**Describe the bug** I have a consumer group, which consists of pods of my service. Pods appear and disappear depending on the load. At some point, my pod appeared and...
**Describe the bug** Hi all, We use `Aivens`'s kafka servers and `aiokafka` for the client side. When getting `NotLeaderForPartitionError` error (documented [here](https://aiven.io/docs/products/kafka/concepts/non-leader-for-partition)), although the error is define `invalid_metadata = True`...
I have two instances of aiokafka producers with `enable_idempotence=True`. Each instance is producing messages using only `send_and_wait` method. Once in a while they start to throw `OutOfOrderSequenceNumber` and producing messages...
Hi, We're using OAUTHBEARER sasl_mechanism to connect to AWS MSK with MSKAuthTokenProvider generated tokens. The tokens have an expiration duration. Is there any way to make the Kafka clients refresh...
When a consumer connection breaks, aiokafka goes into a loop trying to reconnect every 1ms, generating a mountain of logs. This PR introduces a delay between connection attempts with the...
Hi! I want to close both a producer and consumer when Kafka server isn't responding anymore. If I do ```python await producer.stop() ``` ```python await consumer.stop() ``` It will try...
