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
muslpy 3.10–3.910 runs
build_error
glibcpy 3.10–3.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.fixRemove 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.fixUpgrade 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.fixAlways 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`.fixInstall 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.fixEnsure 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.fixEnsure 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.
fixInstall 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.
fixEnsure 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.
fixAdjust 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.
fixEnsure 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.
fixRemove 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.