aiokafka

aiokafka is a Python client for Apache Kafka built directly on top of asyncio, giving Python applications non-blocking access to Kafka’s producer and consumer protocols without wrapping a separate thread pool or subprocess. It exposes AIOKafkaProducer and AIOKafkaConsumer classes that mirror the semantics of the official Java client while fitting naturally into async/await code.

The library implements Kafka’s group coordination protocol directly, so consumers can join a consumer group, get partitions assigned, and commit offsets automatically or manually. It supports SSL, SASL authentication, gzip/snappy/lz4/zstd compression, and idempotent and transactional producers, making it usable for both simple pub/sub pipelines and exactly-once processing workloads in async services.

What You Get

• AIOKafkaProducer for asynchronous, batched, and optionally idempotent/transactional message production
• AIOKafkaConsumer with automatic consumer-group coordination, partition assignment, and offset management
• Built-in support for SSL/TLS and SASL (PLAIN, SCRAM, GSSAPI, OAUTHBEARER) authentication
• Pluggable compression codecs (gzip, snappy, lz4, zstd) for producer and consumer records
• An async admin client for creating/deleting topics and managing consumer group metadata

Common Use Cases

• Building async microservices that publish domain events to Kafka topics
• Consuming Kafka streams inside FastAPI, aiohttp, or other asyncio-based web services
• Implementing exactly-once processing pipelines with idempotent or transactional producers
• Bridging Kafka topics into async task queues or event-driven pipelines

Under The Hood

Architecture aiokafka is organized around client.py (the low-level AIOKafkaClient that manages broker connections and metadata), producer/ and consumer/ packages that build the public AIOKafkaProducer/AIOKafkaConsumer APIs on top of it, and coordinator/ which implements the consumer-group join/sync/heartbeat protocol against a Kafka group coordinator broker. conn.py owns the actual asyncio socket connections and request/response framing, protocol/ encodes/decodes Kafka’s binary wire protocol, and record/ handles Kafka’s record-batch formats (v0-v2) including compression. A Cython extension accelerates record (de)serialization on the hot path.

Tech Stack Pure Python 3.10+ with a small Cython layer for performance-critical record encoding/decoding (compiled via setuptools+Cython at build time). Runtime dependencies are minimal — async-timeout, packaging, and typing_extensions — with optional extras (cramjam for lz4/snappy/zstd compression, gssapi for Kerberos auth) kept out of the default install. Tooling is Ruff for linting/formatting and mypy for static typing.

Code Quality The tests/ directory mirrors the package layout with dedicated suites for the client, consumer, producer, coordinator, and record formats, plus a _testutil.py harness for spinning up Kafka via Docker in CI. Tests run through pytest with asyncio_mode = auto and filterwarnings = error, so warnings fail the suite. mypy is configured with disallow_untyped_defs and disallow_any_generics, indicating the maintainers hold the public API to a typed standard even though full annotation coverage is still a stated TODO in pyproject.toml.

API Design The producer/consumer split follows the same mental model as the official Kafka clients (Java, kafka-python), so the learning curve for anyone who has used a Kafka client before is low. Usage is idiomatic asyncio: await producer.start(), await producer.send_and_wait(...), await producer.stop(), and async for msg in consumer: for streaming reads — no callback registration or manual polling loop is required, which is a meaningful ergonomic improvement over lower-level Kafka clients.