Registry / database / aiokafka

aiokafka

JSON →
library0.14.0pypypi✓ verified 25d ago

aiokafka is an asyncio-compatible client for Apache Kafka, enabling asynchronous production and consumption of messages. It leverages Python's `async/await` syntax for non-blocking I/O operations, making it suitable for high-performance applications. The current version is 0.13.0, and it generally releases new versions every few months, often incorporating new Kafka KIPs and Python version support updates.

pip install aiokafka
INSTALL
IMPORT
SIG · AIOKAFKA
A
aiokafka
databasepythonv0.14.0
Install
2.1s avg
Import
309ms
Disk
26MB
Pass rate
5/ 10
Env Coverage5 / 10
glibc
3.93.13
musl
3.93.13
Install & Compatibility
Where this runs
tested against v0.14.0 · pip install
no network on importno background threads
Install × environment matrix
Each cell = how many times install + import succeeded across repeated harness runs. Partial = flaky.
glibc = Debian/Ubuntu slim · musl = Alpine Linux
musl
py 3.103.910 runs
build_error
glibc
py 3.103.910 runs
installs and imports cleanly · install 2.1s · import 0.309s · 28MB
26MB installed
● package 26MB
Code
Verified usage

Verified import paths — ran on the pinned version, not inferred.

AIOKafkaConsumer
from aiokafka import AIOKafkaConsumer
AIOKafkaProducer
from aiokafka import AIOKafkaProducer
AIOKafkaAdminClient
from aiokafka import AIOKafkaAdminClient
KafkaError
from aiokafka.errors import KafkaError
from aiokafka import KafkaError
KafkaError and other specific error classes are located in the `aiokafka.errors` submodule.

This quickstart demonstrates setting up an `AIOKafkaProducer` to send a message and an `AIOKafkaConsumer` to receive it. Ensure you have a Kafka broker running (e.g., via Docker) and optionally set `KAFKA_BOOTSTRAP_SERVERS` and `KAFKA_TOPIC` environment variables. It highlights the `start()` and `stop()` methods, crucial for managing the client lifecycle in an asyncio application.

import asyncio import os from aiokafka import AIOKafkaConsumer, AIOKafkaProducer BOOTSTRAP_SERVERS = os.environ.get('KAFKA_BOOTSTRAP_SERVERS', 'localhost:9092') TOPIC = os.environ.get('KAFKA_TOPIC', 'my_test_topic') async def main(): producer = AIOKafkaProducer(bootstrap_servers=BOOTSTRAP_SERVERS) consumer = AIOKafkaConsumer( TOPIC, bootstrap_servers=BOOTSTRAP_SERVERS, group_id="my_consumer_group", auto_offset_reset="earliest" ) print(f"Connecting to Kafka at {BOOTSTRAP_SERVERS}") await producer.start() await consumer.start() try: # Produce message print(f"Producing message to topic {TOPIC}") await producer.send_and_wait(TOPIC, b"Hello aiokafka!") print("Message produced.") # Consume message print(f"Consuming message from topic {TOPIC}") async for msg in consumer: print(f"Consumed: offset={msg.offset}, key={msg.key}, value={msg.value.decode()}") break # Only consume one message for the example finally: print("Stopping producer and consumer...") await producer.stop() await consumer.stop() print("Stopped.") if __name__ == '__main__': asyncio.run(main())
Debug
Known issues
breakingThe `api_version` parameter has been removed from `AIOKafkaConsumer`, `AIOKafkaProducer`, and `AIOKafkaAdminClient` constructors. API versions are now resolved automatically at connection time.
fix
Remove the `api_version` parameter from your client constructor calls. The library will negotiate the correct API version with the Kafka brokers.
affects: >=0.13.0
breakingaiokafka has progressively dropped support for older Python versions. Ensure your environment meets the minimum requirement.
fix
Upgrade your Python environment to 3.10 or newer if you are using aiokafka 0.12.0+.
affects: 0.12.0 drops Python 3.8; 0.8.1 drops Python 3.7; 0.8.0 drops Python 3.6. Current minimum is Python 3.10.
gotchaaiokafka is an asyncio-native library. All operations that interact with Kafka are `await`-able. Using it outside an `async` context or without `await` will lead to runtime errors or incorrect behavior.
fix
Always use `await` with `start()`, `stop()`, `send_and_wait()`, and when iterating over the consumer (e.g., `async for msg in consumer:`). Ensure your code runs within an `asyncio` event loop.
affects: All versions
gotchaFor Kafka messages compressed with Snappy, LZ4, or ZStandard, the corresponding Python compression libraries must be installed as extra dependencies, otherwise, messages compressed with these codecs cannot be decompressed, leading to `ImportError` or `CompressionError`.
fix
Install aiokafka with the necessary compression extras: `pip install aiokafka[snappy,lz4,zstd]` for full support.
affects: All versions
gotchaA `KafkaConnectionError` (e.g., `[Errno 111] Connect call failed`) indicates that the AIOKafka client was unable to connect to the specified Kafka broker(s). This usually means the Kafka server is not running, is inaccessible from the client's network, or is listening on a different address/port.
fix
Ensure your Kafka broker is running and accessible from where the client application is executed. Verify the broker's advertised listeners are correctly configured and reachable on the specified host and port (e.g., `localhost:9092`). If running in Docker, ensure proper network configuration between containers or with the host.
affects: All versions
gotchaaiokafka includes C extensions that require a C compiler (like GCC) and Python development headers for installation. In minimal environments (e.g., Alpine Linux Docker images), these build dependencies are often missing, leading to installation failures.
fix
Ensure your environment has the necessary build tools. For Alpine Linux, install `build-base` and `python3-dev` (e.g., `apk add build-base python3-dev`). For Debian/Ubuntu, install `build-essential` and `python3-dev` (e.g., `apt-get install build-essential python3-dev`).
affects: All versions
Errors
Common errors & fixes
ModuleNotFoundError: No module named 'aiokafka'
The 'aiokafka' library is not installed in the Python environment.
fix
Install the 'aiokafka' library using pip: 'pip install aiokafka'.
aiokafka.errors.ProducerClosed: ProducerClosed
Attempting to send a message using an 'AIOKafkaProducer' instance that has already been closed.
fix
Ensure the producer is started before sending messages and not closed prematurely. Use 'await producer.start()' before sending messages and 'await producer.stop()' after all messages are sent.
UnknownMemberIdError
The consumer is not recognized as a member of the consumer group, possibly due to session timeouts or rebalancing.
fix
Adjust consumer configuration parameters like 'session.timeout.ms' and 'heartbeat.interval.ms' to appropriate values to maintain group membership.
TypeError: 'CreateTopicsRequest_v1' object has no attribute 'build_request_header'
Incompatibility between 'faust' and 'aiokafka' versions, leading to missing attributes.
fix
Ensure that both 'faust' and 'aiokafka' libraries are updated to compatible versions.
DeprecationWarning: The loop argument is deprecated
Explicitly passing the `loop` argument to `AIOKafkaConsumer` or `AIOKafkaProducer` constructors is deprecated in newer `aiokafka` and `asyncio` versions. In modern `asyncio`, the event loop is usually implicitly managed.
fix
Remove the `loop` argument when initializing `AIOKafkaConsumer` or `AIOKafkaProducer` (e.g., `consumer = AIOKafkaConsumer('topic')` instead of `consumer = AIOKafkaConsumer('topic', loop=asyncio.get_event_loop())`). Ensure the client is started and stopped within an `async` context using `await`.
Upgrade
Version history
0.14.0latest on PyPI · released Apr 29, 2026
Audit
Dependencies
lz4optionalRequired for LZ4 compression codec support in Kafka.
zstandardoptionalRequired for ZStandard compression codec support in Kafka.
python-snappyoptionalRequired for Snappy compression codec support in Kafka.
Agent activity
59 hits · last 30 days
node
52
OpenAI (training)
2
Resources
aiokafka — pip install aiokafka · libregistry