aiokafka
aiokafka copied to clipboard
aio-libs
asyncio client for kafka
```python consumer_ = aiokafka.AIOKafkaConsumer( *topics, bootstrap_servers=f"192.168.1.100:9092", ) await consumer_.start() try: async for msg in consumer_: print(msg) finally: await consumer_.stop() ``` `aiokafka` outputs a lot of error logs when I disconnect...
I have been working to build and test this package for amd64 and arm64 architectures. While testing this package with “make ci-test-all” getting errors, it looks like DOCKER_IMAGE “aiolibs/kafka” is...
**Describe the bug** Hi, Not sure if a bug or an implementation issue on my side, but: My goal is to have a consumer that yields messages and records each...
### Changes Fixes #680 This is a rebase of https://github.com/aio-libs/aiokafka/pull/682 -- see https://github.com/aio-libs/aiokafka/pull/682#issuecomment-1070337629 ### Checklist - [ ] I think the code is well written - [ ] Unit tests...
### Changes Fixes # Implements [KIP-345](https://cwiki.apache.org/confluence/display/KAFKA/KIP-345%3A+Introduce+static+membership+protocol+to+reduce+consumer+rebalances) in aiokafka ### Checklist - [x] I think the code is well written - [ ] Unit tests for the changes exist - [x]...
Calling `getmany` seems to [suggest it updates consumed offsets at the same time](https://github.com/aio-libs/aiokafka/blob/081087d1c92e2bf6ee3660843f641bfb4f3fd754/aiokafka/consumer/fetcher.py#L1043-L1045). There are warnings in various places around manual commits, and suggestions that committing after each message is...
ConsumerRebalanceListener.on_partitions_assigned docstring says: This method will be called *after* partition re-assignment completes and *before* the consumer starts fetching data again. I assume this follows original Java API description: ConsumerRebalanceListener.onPartitionsAssigned This...
### Update [black](https://pypi.org/project/black) from **21.10b0** to **22.1.0**. Changelog ### 22.1.0 ``` At long last, _Black_ is no longer a beta product! This is the first non-beta release and the first...
Should I catch UnknownProducerId or BrokerResponseError in my code? https://github.com/aio-libs/aiokafka/blob/master/aiokafka/errors.py#L318
I am using Python 3.9.9. I am facing the problem of message duplicate even after a successful commit. The same message appears multiple times even after committing. I can see...
