confluent-kafka
confluent-kafka is Confluent's Python client for Apache Kafka. It is a thin binding over librdkafka, the C client that also backs the Go, .NET and C++ clients, so the actual networking, batching, compression, retries and partition assignment happen in C while Python holds a handle. You get four main objects: Producer for writing, Consumer for reading in a group, AdminClient for topics, ACLs and configs, and SchemaRegistryClient with Avro, Protobuf and JSON Schema serializers for the Confluent Schema Registry. Configuration is a plain dict of librdkafka's dotted string keys such as bootstrap.servers, group.id and enable.auto.commit, which means the client's real reference manual is librdkafka's configuration list rather than anything Pythonic. It works against any Kafka broker from 0.8 onward, not just Confluent's.
The client to use if Kafka is load-bearing in your system: it is fast, it implements the whole protocol, and it behaves the same as Confluent's other clients. Budget for the fact that you are operating a C library through a dict of string config keys, and check the Kerberos wheel gap before you promise anything to your security team.
Use it if
- You need throughput: batching, compression and the network loop run in C, so a single process sustains rates a pure-Python client cannot reach
- You need the parts of Kafka that only a full client implements: transactions and exactly-once read-process-write, cooperative rebalancing, idempotent produce, consumer group metadata
- You use Confluent Schema Registry and want Avro, Protobuf or JSON Schema serializers that register and resolve schemas for you rather than hand-rolling the wire format
- You already run other Confluent clients: sharing librdkafka means the same config keys and the same behaviour across your Go and Python services
- You are on Confluent Cloud, where this is the client the platform's own docs, config snippets and support process assume
- You need SASL Kerberos or GSSAPI: the prebuilt Linux wheels are compiled without it, so you are building from source against your system librdkafka and its Kerberos libraries, in every image you ship
- You want a Pythonic API: this is a C client wearing a Python coat, so config is a dict of dotted strings that fails at construction on a typo, errors arrive as KafkaError objects you inspect rather than exceptions you catch, and delivery callbacks only fire when you remember to call poll()
- Your application is async-first and you need every feature: the AIOProducer and AIOConsumer are recent, and the batched async produce path does not support per-message headers, so header-carrying writes fall back to the sync producer in a thread
- You want stream processing rather than a client: there are no joins, windows or state stores here, and Quix Streams, Faust or Flink is what that job needs
- You need to debug behaviour you cannot see: when something goes wrong the answer is usually in librdkafka's configuration reference and its debug=broker,topic,msg output, which is a C library's documentation and log format, not Python tracebacks
- You only need a work queue between two services: running Kafka to get one is a large operational bill compared with Redis streams, SQS or a database table
Setup reality
pip install confluent-kafka gets a prebuilt wheel with librdkafka statically linked on mainstream Linux, macOS and Windows, so there is usually nothing to compile. Two things bite. First, the wheels omit SASL Kerberos/GSSAPI support entirely; needing it means an apt-get of librdkafka-dev plus a source build, and that has to be reproduced in your Docker image. Second, Schema Registry is not included: you install extras such as confluent-kafka[avro,schemaregistry] or [protobuf,schemaregistry], and the Data Contract rules and field-level encryption features need [rules], which pulls a large tree including cloud KMS SDKs. Python 3.8 or newer. Once installed, the API expects you to drive it. Producer.produce() only enqueues, so you must call poll(0) in your loop and flush() before exit or you lose buffered messages and never see delivery callbacks; a full queue raises BufferError rather than blocking. Consumer.poll() returns a message whose error() you have to check before touching value(), and a Consumer that is not close()d leaves the group to time out instead of triggering an immediate rebalance. Producer, Consumer and AdminClient are thread safe; the preview ShareConsumer is not.
Patterns
Produce with delivery reportsproduce-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 pendingproduce() only enqueues. Without the poll(0) inside the loop, callbacks never run; without flush() before exit, buffered messages are dropped silently. flush returns the count still undelivered, so check it rather than assuming success.
Cope with a full producer queuehandle-queue-full
while True:
try:
p.produce("orders", value=payload, callback=on_delivery)
break
except BufferError:
p.poll(0.5) # drain space, then retryBufferError means queue.buffering.max.messages is reached, usually because the broker is slow or unreachable. Raising the limit hides the problem and grows memory; poll-and-retry applies backpressure to your own producer instead.
Consume in a group and shut down cleanlyconsume-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()Always check msg.error() before msg.value(); errors arrive as messages, not exceptions. Skipping close() means the group waits out session.timeout.ms before rebalancing, so a rolling deploy stalls consumption for that long.
Commit offsets only after the work succeedsmanual-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 the default enable.auto.commit=True the offset advances on a timer whether or not your handler succeeded, so a crash loses messages. Committing a message commits offset+1 for that partition, which is what you want.
React to partition assignment changesrebalance-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)on_revoke is your only chance to commit before the partitions move. With partition.assignment.strategy=cooperative-sticky the callbacks receive only the partitions that changed, not the whole set, so do not treat the argument as the full assignment.
Connect over SASL_SSLconnect-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",
}The key is sasl.mechanisms (plural), and an unknown or misspelled key raises KafkaException at client construction rather than being ignored. These wheels have no Kerberos support, so sasl.mechanisms=GSSAPI needs a source build.
Produce Avro through Schema Registryavro-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 confluent-kafka[avro,schemaregistry] first; the base package has none of this. The SerializationContext matters because the subject name strategy derives the registry subject from the topic and whether it is the key or the value.
Consume, transform, produce in one transactionexactly-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()transactional.id must be stable and unique per producer instance, so it cannot be a random UUID per restart or fencing stops working. The reading side needs isolation.level=read_committed or it sees aborted messages.
Produce from an event loopasyncio-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())There are two awaits: the first enqueues and hands back a future, the second waits for the broker acknowledgement. The batched async path does not carry per-message headers, so header-bearing writes need the sync Producer via run_in_executor.
Create topics from codeadmin-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)Every AdminClient call returns a dict of concurrent futures, so nothing has happened until you call result(). Partition count can be increased later but never decreased, and increasing it changes which partition a key hashes to.
Read a specific offset rangeseek-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() + 1assign() bypasses the consumer group's partition assignment, so use it for backfills, not for scaled-out consumers. Calling seek() right after subscribe() fails because no partitions are assigned yet; do it from the on_assign callback.
Trade latency for throughput, and watch what happenstune-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 above zero is the single biggest throughput lever, at the cost of that much added latency per message. enable.idempotence pins acks=all and bounded in-flight requests, so setting acks=1 alongside it raises a config error instead of quietly winning.
Alternatives
| Package | Registry | Pick it when |
|---|---|---|
| aiokafka | PyPI | Your service is built on asyncio and you want a native async client with no C extension to build |
| kafka-python | PyPI | You need a pure-Python client that installs anywhere, and your volumes are low enough that throughput is not the constraint |
| quixstreams | PyPI | You want stateful stream processing (windows, aggregations, joins) rather than a raw producer and consumer |
| faust-streaming | PyPI | You want a Kafka Streams-style application framework in Python with agents and tables |