kafka-python review
kafka-python 3.0.11 is a pure-Python Apache Kafka client with producer, consumer, admin, protocol, and command-line interfaces. Version 3 rebuilt networking around a dedicated I/O thread, generates protocol classes from Apache Kafka JSON schemas, enables producer idempotence and `acks='all'` by default, and expands cooperative rebalancing, transactions, and admin operations. The current 3.0.11 fix corrects JSON deserialization when `JsonSerializer` receives null data. Our package inspection found no `py.typed`, so runtime capability is broader than its static typing contract.
kafka-python 3.0.11 installed in 0.3 seconds as a single 4 MB package and imported in 0.54 seconds in our sandbox, with 0 pip-audit findings; it is the practical Kafka client when a pure-Python deployment matters. Pick `confluent-kafka` for native throughput or vendor integrations, and pick `aiokafka` when the application API must be asyncio-native.
We installed it
| Install | ✓ · 0.3s | 1 package on disk · 4 MB |
| Import | ✓ | import kafka in 0.54s · pure Python · requires Python >=3.8 |
| Known vulns | 0 | (pip-audit) |
Answers from our run
Does kafka-python install cleanly?
Yes. In a fresh container with an empty cache, pip install kafka-python finished in 0.3s, leaving 1 package and 4 MB on disk. pip-audit reported no known vulnerabilities.
What does kafka-python need to run?
Python >=3.8, and nothing compiled: it is pure Python. In our run import kafka succeeded in 0.54s.
kafka-python or confluent-kafka: which should you use?
confluent-kafka: You want librdkafka speed, Confluent services, Schema Registry support, or vendor-backed client behavior. kafka-python 3.0.11 installed in 0.3 seconds as a single 4 MB package and imported in 0.54 seconds in our sandbox, with 0 pip-audit findings; it is the practical Kafka client when a pure-Python deployment matters.
When should you not use kafka-python?
Your throughput or CPU budget calls for a native client: the README recommends optional native crc32c because checksum work otherwise stays in Python
Use it if
- You need Kafka producers, consumers, and admin calls in an environment where a pure-Python client is easier to ship than librdkafka
- You want a client whose high-level API follows the Java Kafka client closely and whose protocol code is inspectable in Python
- You need Kafka command-line operations on a machine without the broker's Java scripts
- Your workload values portability and broker feature coverage more than the highest possible message throughput
- Your throughput or CPU budget calls for a native client: the README recommends optional native crc32c because checksum work otherwise stays in Python
- Your service needs public asyncio producer and consumer APIs: version 3 uses async syntax inside its own I/O layer, but the documented high-level clients remain synchronous
- You need bundled static typing: our inspection found no `py.typed`, even though generated protocol classes include internal stub work
- You expect `flush()` to confirm delivery: the producer documentation says it only gets pending records onto the network and does not prove success
- You want vendor-backed Schema Registry tooling and librdkafka configuration parity: `confluent-kafka` is built for that stack
- You plan to share one consumer between worker threads: the API documentation states that `KafkaConsumer` is not thread-safe
Setup reality
Our fresh Python 3.12 install of kafka-python 3.0.11 completed in 0.3 seconds. It left 1 package and 4 MB on disk, reported 8 direct dependencies, imported as kafka in 0.54 seconds, and returned 0 known vulnerabilities from pip-audit. The distribution is pure Python, requires Python 3.8 or newer, does not ship py.typed, and did not expose a license value in the package metadata we measured.
A client still needs reachable broker addresses and security settings. Supply several bootstrap_servers entries because those addresses only seed cluster discovery. The default protocol is PLAINTEXT; hosted clusters usually need SASL_SSL or SSL plus a CA, mechanism, username and password, or a token provider. Keep credentials outside source. Gzip uses the standard library, while LZ4, Snappy, Zstandard, and optimized CRC32C require optional packages.
KafkaProducer.send() queues work and returns a future. Wait on future.get() or register callbacks when the result matters because flush() alone does not establish delivery. Version 3 defaults to idempotence with acks='all' and unlimited retries, but conflicting manual values can disable those defaults with a warning. Set enable_idempotence=True explicitly when misconfiguration should fail. A transactional producer also needs a stable, unique transactional_id, and readers must use READ_COMMITTED to hide aborted records.
Consumers yield byte keys and values unless serializers are configured. group_id=None means no group commits, auto_offset_reset='latest' skips older records when no committed offset exists, and iteration waits indefinitely unless consumer_timeout_ms is set. Auto-commit runs every 5 seconds by default and may record progress before your side effect finishes. Disable it for at-least-once work, commit after success, and keep processing beneath max_poll_interval_ms or expect a rebalance. Do not share a consumer across threads.
Patterns
Send a record and wait for the broker result produce-bytes
from kafka import KafkaProducer
producer = KafkaProducer(bootstrap_servers=["kafka-1:9092", "kafka-2:9092"])
future = producer.send("events", key=b"user-42", value=b"signed-in")
metadata = future.get(timeout=30)
producer.close()`send()` returns before delivery; `future.get()` exposes broker errors and returns the assigned topic, partition, and offset.
Serialize dictionaries as JSON produce-json
from kafka import JsonSerializer, KafkaProducer
producer = KafkaProducer(
bootstrap_servers="localhost:9092",
value_serializer=JsonSerializer(),
)
producer.send("events", {"kind": "signup", "user_id": 42}).get(timeout=30)Version 3.0.11 fixes null handling in `JsonSerializer`; consumers need a compatible deserializer rather than assuming decoded dictionaries.
Consume decoded JSON records consume-json
from kafka import JsonSerializer, KafkaConsumer
consumer = KafkaConsumer(
"events",
bootstrap_servers="localhost:9092",
group_id="analytics",
value_deserializer=JsonSerializer(),
)
for record in consumer:
print(record.partition, record.offset, record.value)Without a deserializer, keys and values are bytes; iteration can wait forever unless you set `consumer_timeout_ms`.
Commit an offset after the side effect commit-after-processing
consumer = KafkaConsumer(
"jobs",
bootstrap_servers="localhost:9092",
group_id="workers",
enable_auto_commit=False,
auto_offset_reset="earliest",
)
for record in consumer:
process(record.value)
consumer.commit()Manual commit after `process` gives at-least-once handling; a crash between the side effect and commit can repeat the record.
Stop iteration after an idle interval limit-consumer-wait
consumer = KafkaConsumer(
"audit-events",
bootstrap_servers="localhost:9092",
consumer_timeout_ms=1000,
)
for record in consumer:
print(record.value)
consumer.close()The default consumer iterator blocks indefinitely; `consumer_timeout_ms=1000` exits after 1 second without records.
Read one partition without group assignment assign-partition
from kafka import KafkaConsumer, TopicPartition
consumer = KafkaConsumer(bootstrap_servers="localhost:9092")
partition = TopicPartition("events", 2)
consumer.assign([partition])
consumer.seek(partition, 0)
record = next(consumer)Manual `assign()` bypasses group partition management and cannot be combined with `subscribe()` on the same consumer.
Connect to a SASL over TLS cluster configure-sasl-ssl
import os
from kafka import KafkaProducer
producer = KafkaProducer(
bootstrap_servers=["broker.example.com:9093"],
security_protocol="SASL_SSL",
sasl_mechanism="SCRAM-SHA-512",
sasl_plain_username=os.environ["KAFKA_USERNAME"],
sasl_plain_password=os.environ["KAFKA_PASSWORD"],
ssl_cafile="/etc/ssl/certs/cluster-ca.pem",
)PLAINTEXT is the default; use broker hostnames covered by the certificate and keep SASL secrets outside the codebase.
Require idempotent producer settings enable-idempotence
producer = KafkaProducer(
bootstrap_servers="localhost:9092",
enable_idempotence=True,
acks="all",
)Version 3 enables idempotence by default, but setting it explicitly turns conflicting reliability options into configuration errors instead of warnings.
Write a Kafka transaction run-transaction
producer = KafkaProducer(
bootstrap_servers="localhost:9092",
transactional_id="billing-writer-01",
)
producer.init_transactions()
producer.begin_transaction()
try:
producer.send("ledger", value=b"charge:42").get(timeout=30)
producer.commit_transaction()
except Exception:
producer.abort_transaction()
raiseEach live producer instance needs a stable unique `transactional_id`; consumers must request `READ_COMMITTED` to exclude aborted records.
Create a topic through the admin client create-topic
from kafka.admin import KafkaAdminClient
admin = KafkaAdminClient(bootstrap_servers="localhost:9092")
admin.create_topics({
"events-v1": {
"num_partitions": 6,
"replication_factor": 3,
}
})
admin.close()Version 3 deprecates `NewTopic` lists in favor of names mapped to option dictionaries; broker authorization still controls topic creation.
Expose client network events enable-debug-logging
import logging
logging.basicConfig(level=logging.INFO)
logging.getLogger("kafka").setLevel(logging.DEBUG)Debug logs reveal bootstrap, metadata, coordinator, and transport events; review them for broker addresses and other deployment details before sharing.
Inspect a cluster without Kafka's Java scripts use-admin-cli
kafka-python admin -b kafka-1:9092,kafka-2:9092 cluster describeThe package installs admin, producer, and consumer CLI entry points; pass the same authentication and TLS configuration required by the Python clients.
Alternatives
| Package | Registry | Pick it when |
|---|---|---|
| confluent-kafka | PyPI | You want librdkafka speed, Confluent services, Schema Registry support, or vendor-backed client behavior |
| aiokafka | PyPI | An asyncio service needs awaitable producer and consumer APIs |
| quixstreams | PyPI | You need a higher-level stream-processing layer with state, windows, and dataframe-style operations |
| Faust | PyPI | You are maintaining a Faust actor and table application and accept its framework model |
More data guides
numpy · fsspec · pandas · sqlalchemy · pyarrow · lxml · the whole shelf →
How this guide is made: grounded in the library's documentation, release notes, changelog, and issue history, on a fixed rubric — not a hands-on install of every release. The 50 most-downloaded entries are additionally install-verified in clean containers. Corrections: contact the desk.