confluent-kafka review
confluent-kafka 2.15.0 is a Python interface to librdkafka for Kafka producers, consumers, administration, transactions, and Confluent Schema Registry serializers. Protocol I/O and batching live in the compiled extension while Python supplies configuration and callbacks. Version 2.15.0 adds a preview KIP-932 ShareConsumer with record acknowledgement and redelivery, an AWS IAM OAUTHBEARER extra, and Schema Registry pool mapping. Our sandbox imported the typed extension in 0.10 seconds. Confluent explicitly keeps ShareConsumer out of production for now because its interface can change and several broker and client features are missing.
confluent-kafka 2.15.0 installed in 0.3 seconds and imported in 0.10 seconds in our sandbox, with 0 audit findings and a 14 MB compiled footprint. It is the practical Python choice when Kafka is already core infrastructure, but Kerberos wheel limits, callback polling, and the preview ShareConsumer deserve explicit design decisions.
We installed it
| Install | ✓ · 0.3s | 1 package on disk · 14 MB |
| Import | ✓ | import confluent_kafka in 0.10s · compiled extensions · py.typed · requires Python >=3.8 |
| Known vulns | 0 | (pip-audit) |
Answers from our run
Does confluent-kafka install cleanly?
Yes. In a fresh container with an empty cache, pip install confluent-kafka finished in 0.3s, leaving 1 package and 14 MB on disk. pip-audit reported no known vulnerabilities.
What does confluent-kafka need to run?
Python >=3.8, and a platform wheel with compiled extensions. In our run import confluent_kafka succeeded in 0.10s, and the package ships py.typed for type checkers.
confluent-kafka or aiokafka: which should you use?
aiokafka: Choose it when an asyncio-native Python Kafka API matters more than using librdkafka. confluent-kafka 2.15.0 installed in 0.3 seconds and imported in 0.10 seconds in our sandbox, with 0 audit findings and a 14 MB compiled footprint.
When should you not use confluent-kafka?
You need windows, joins, tables, or stateful stream operators. This package is a Kafka client, so those processing concepts remain application work.
Use it if
- A Python service needs Kafka producer, consumer-group, or AdminClient access through librdkafka.
- You need idempotent writes, transactions, manual offset control, or cooperative group assignment in one client.
- Avro, Protobuf, or JSON Schema messages already use Confluent Schema Registry.
- An asyncio service can use AIOProducer or AIOConsumer and does not require headers on the batched producer path.
- You need windows, joins, tables, or stateful stream operators. This package is a Kafka client, so those processing concepts remain application work.
- Kerberos must work from the stock Linux wheel. The install guide says those wheels omit SASL GSSAPI support.
- Every asyncio-produced record needs headers. The AIOProducer documentation says its batched path cannot attach per-message headers.
- Your team wants Python attributes to validate client settings. Most controls are librdkafka string keys, and misspelled or conflicting values often surface when the client starts.
- You are considering ShareConsumer for production. In 2.15.0 it is preview-only, synchronous, single-threaded, lacks lock renewal, and cannot subscribe with a regex.
- The workload is occasional background jobs with no existing Kafka cluster. Running Kafka is a much larger commitment than installing this client.
Setup reality
We installed confluent-kafka 2.15.0 in a clean Python 3.12 Bookworm sandbox in 0.3 seconds. It left one package and 14 MB on disk. pip-audit found 0 known vulnerabilities, and import confluent_kafka completed in 0.10 seconds. The distribution requires Python 3.8 or newer, contains compiled .so files, and includes py.typed. Our package measurement reported 312 direct dependencies.
The first client still needs bootstrap.servers. TLS or SASL clusters add protocol, mechanism, username, password, and certificate settings. Confluent Cloud keeps Kafka and Schema Registry credentials separate; the README traces Schema Registry 401 or 403 responses to using the Kafka key in basic.auth.user.info. Serializer support and AWS IAM OAUTHBEARER come from optional extras, so pin those extras in deployment requirements.
Producer.produce() places a record in a local queue. Delivery callbacks run only while poll() or flush() serves them, and a saturated queue raises BufferError. During shutdown, inspect the value returned by flush() because it is the count still undelivered. A consumer poll can yield None, data, or a message with error() set. Always close the consumer so the group does not wait for its session timeout.
Producer, Consumer, and AdminClient are documented as thread safe. ShareConsumer is single-threaded. In AIOProducer, the first await enqueues and returns a delivery future; the second await observes the broker result. ShareConsumer 2.15.0 needs Kafka 4.2.0 share groups, and its explicit mode requires every record to be acknowledged before the next poll.
Patterns
Queue records and collect delivery reports produce-messages
from confluent_kafka import Producer
p = Producer({"bootstrap.servers": "broker1:9092,broker2:9092"})
def on_delivery(err, msg):
if err is not None:
print("delivery failed:", err)
else:
print("delivered to", msg.topic(), msg.partition(), msg.offset())
for row in rows:
p.poll(0) # serve delivery callbacks from earlier produces
p.produce("orders", key=row.id.encode(), value=row.to_json().encode(),
callback=on_delivery)
p.flush(30) # blocks until the queue drains, returns messages still pending`produce()` only queues locally. Calling `poll()` serves earlier callbacks, while `flush(30)` returns the number of records still pending after 30 seconds.
Wait for space when the producer queue fills handle-queue-full
while True:
try:
p.produce("orders", value=payload, callback=on_delivery)
break
except BufferError:
p.poll(0.5) # drain space, then retry`BufferError` means the local queue reached its configured limit. Polling lets completed deliveries free slots before the retry.
Consume records without losing error messages consume-loop
from confluent_kafka import Consumer, KafkaError, KafkaException
c = Consumer({
"bootstrap.servers": "broker1:9092",
"group.id": "order-indexer",
"auto.offset.reset": "earliest",
})
c.subscribe(["orders"])
try:
while running:
msg = c.poll(1.0)
if msg is None:
continue
if msg.error():
if msg.error().code() == KafkaError._PARTITION_EOF:
continue
raise KafkaException(msg.error())
handle(msg.value())
finally:
c.close()A consumer can return an error-bearing message instead of raising. `close()` leaves the group cleanly and shortens rebalancing during a deployment.
Commit offsets after the work succeeds manual-commit
c = Consumer({
"bootstrap.servers": "broker1:9092",
"group.id": "order-indexer",
"enable.auto.commit": False,
})
msg = c.poll(1.0)
if msg and not msg.error():
index(msg.value()) # do the work first
c.commit(msg, asynchronous=False)With auto commit disabled, commit only after the side effect finishes. A synchronous commit waits for the broker, so batching several records usually costs less.
Finish partition work during rebalancing rebalance-callbacks
def on_assign(consumer, partitions):
print("assigned", [(p.topic, p.partition) for p in partitions])
def on_revoke(consumer, partitions):
flush_local_state()
consumer.commit(asynchronous=False)
c.subscribe(["orders"], on_assign=on_assign, on_revoke=on_revoke)Cooperative assignment can revoke only part of the current set. Persist and commit state for the partitions named in the callback.
Connect with SASL over TLS connect-confluent-cloud
conf = {
"bootstrap.servers": "pkc-xxxxx.region.provider.confluent.cloud:9092",
"security.protocol": "SASL_SSL",
"sasl.mechanisms": "PLAIN",
"sasl.username": os.environ["KAFKA_KEY"],
"sasl.password": os.environ["KAFKA_SECRET"],
"client.id": "order-indexer-1",
}Keep these secrets outside source control. Stock Linux wheels omit GSSAPI, so Kerberos requires the documented source-build route and a compatible librdkafka.
Encode Avro through Schema Registry avro-schema-registry
from confluent_kafka import Producer
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer
from confluent_kafka.serialization import SerializationContext, MessageField
sr = SchemaRegistryClient({"url": "http://localhost:8081"})
to_dict = lambda user, ctx: {"name": user.name, "email": user.email}
serializer = AvroSerializer(sr, user_schema_str, to_dict)
p = Producer({"bootstrap.servers": "broker1:9092"})
p.produce("users", key="u1",
value=serializer(user, SerializationContext("users", MessageField.VALUE)))
p.flush()Install the Avro and Schema Registry extras. The serialization context controls topic plus key or value subject naming, which becomes part of compatibility.
Write output and offsets transactionally exactly-once-transactions
producer = Producer({
"bootstrap.servers": "broker1:9092",
"transactional.id": "order-transformer-1",
})
producer.init_transactions()
producer.begin_transaction()
for msg in batch:
producer.produce("orders-clean", value=transform(msg.value()))
producer.send_offsets_to_transaction(
consumer.position(consumer.assignment()),
consumer.consumer_group_metadata(),
)
producer.commit_transaction()Each live producer needs a stable, unique `transactional.id`. Output consumers need `isolation.level=read_committed` to hide aborted records.
Use the two-stage asyncio producer result asyncio-producer
import asyncio
from confluent_kafka.aio import AIOProducer
async def main():
p = AIOProducer({"bootstrap.servers": "broker1:9092"})
try:
delivery_future = await p.produce("orders", value=b"hello")
msg = await delivery_future
await p.flush()
finally:
await p.close()
asyncio.run(main())The first await returns a future after enqueueing. Await that future for delivery; records with headers need the synchronous producer path.
Resolve every AdminClient future admin-create-topics
from confluent_kafka.admin import AdminClient, NewTopic
a = AdminClient({"bootstrap.servers": "broker1:9092"})
topics = [NewTopic("orders", num_partitions=6, replication_factor=3)]
for topic, fut in a.create_topics(topics).items():
try:
fut.result()
print("created", topic)
except Exception as e:
print("failed", topic, e)Administration calls return one future per resource. Calling `result()` is where broker authorization and validation failures become visible.
Replay from a fixed partition offset seek-and-replay
from confluent_kafka import Consumer, TopicPartition
c = Consumer({"bootstrap.servers": "broker1:9092", "group.id": "replay-tool"})
tp = TopicPartition("orders", 0, 15000)
c.assign([tp])
low, high = c.get_watermark_offsets(tp)
while tp.offset < high:
msg = c.poll(1.0)
if msg and not msg.error():
handle(msg.value())
tp.offset = msg.offset() + 1`assign()` bypasses consumer-group assignment and fits repair tools. With `subscribe()`, wait for assignment before seeking.
Tune batching and expose client statistics tune-and-observe
import json
conf = {
"bootstrap.servers": "broker1:9092",
"linger.ms": 20,
"batch.size": 262144,
"compression.type": "lz4",
"enable.idempotence": True,
"statistics.interval.ms": 30000,
"stats_cb": lambda s: report(json.loads(s)),
}`linger.ms` waits for a larger batch. Enabling idempotence constrains acknowledgements and in-flight settings, and conflicting values fail during client creation.
Alternatives
| Package | Registry | Pick it when |
|---|---|---|
| aiokafka | PyPI | Choose it when an asyncio-native Python Kafka API matters more than using librdkafka. |
| kafka-python | PyPI | Choose it when a pure Python client is required and its supported Kafka feature set covers the service. |
| faust-streaming | PyPI | Choose it when the application needs stream agents, tables, and transformations above Kafka consumption. |
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.