confluent-kafka review
confluent-kafka 2.15.1 gives Python code direct access to Kafka through the compiled librdkafka client. It covers producing, consumer groups, broker administration, transactions, and Confluent Schema Registry codecs, with synchronous and asyncio APIs. The 2.15.1 patch changes deserialization order so keys are decoded before values, fixes a Schema Registry subject-prefix parameter, removes race conditions, and repairs a Python 3.14 crash after `AdminClient.delete_records()`. The KIP-932 ShareConsumer introduced in 2.15.0 remains a preview and Confluent says not to run it in production.
confluent-kafka 2.15.0 installed in 0.3 seconds, occupied 14 MB, imported in 0.10 seconds, and produced 0 known vulnerability findings in our sandbox. Install current 2.15.1 when Python must talk to an existing Kafka deployment through librdkafka, but keep the preview ShareConsumer out of production.
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?
Pick aiokafka when an asyncio-first Python API matters more than using librdkafka or Confluent's Schema Registry clients. confluent-kafka 2.15.0 installed in 0.3 seconds, occupied 14 MB, imported in 0.10 seconds, and produced 0 known vulnerability findings in our sandbox.
When should you not use confluent-kafka?
You need joins, windows, tables, or durable stream-processing state. confluent-kafka is a client, so your application still owns that processing layer.
Use it if
- Your Python service needs Kafka transactions, idempotent production, consumer groups, or broker administration through librdkafka.
- Avro, Protobuf, or JSON Schema records must use Confluent Schema Registry from the same client package.
- A FastAPI, aiohttp, or other asyncio service needs nonblocking Kafka producer and consumer calls.
- The application needs delivery callbacks, manual offset control, or client statistics rather than a minimal send-and-receive wrapper.
- You need joins, windows, tables, or durable stream-processing state. confluent-kafka is a client, so your application still owns that processing layer.
- Your deployment policy forbids native code. Our install contained compiled `.so` files, which rules out a pure-Python artifact chain.
- Kerberos must work from the standard Linux wheel. The install notes say those wheels leave out SASL GSSAPI support and direct Kerberos users to a source build.
- Every record sent through the batched asyncio producer needs headers. The README says that path cannot attach per-message headers and points those calls back to the synchronous producer.
- You want KIP-932 queues in production now. ShareConsumer is still preview-only, needs Kafka 4.2.0, has no async client or lock renewal, and accepts exact topic names rather than regex subscriptions.
Setup reality
We installed confluent-kafka 2.15.0 without a cache in an unprivileged Python 3.12 Bookworm container with 3 CPUs and 8 GB of RAM. Installation took 0.3 seconds, left one package using 14 MB, and import confluent_kafka worked in 0.10 seconds. pip-audit found 0 known vulnerabilities. The wheel contained compiled .so files and py.typed. Python 3.8 is the minimum. Its metadata listed 312 direct dependencies, many behind optional extras.
A client cannot start without bootstrap.servers. TLS and SASL deployments also need the matching security protocol, mechanism, and credentials. Confluent Cloud issues separate keys for Kafka and Schema Registry. The troubleshooting guide ties Schema Registry 401 and 403 responses to using a Kafka key in basic.auth.user.info. Avro, Protobuf, JSON Schema, field-level rules, and AWS IAM authentication each require the relevant extra in your pinned requirements.
Producer.produce() writes to a local queue before Kafka confirms anything. Call poll() during normal work so delivery callbacks run, handle BufferError when that queue fills, and read the integer returned by flush() at shutdown because it reports records still queued. A consumer poll may return None, a record, or a record carrying error(). Close consumers explicitly so group departure does not wait on the session timeout.
Producer and AdminClient may be shared across threads, but Consumer and ShareConsumer instances should not be called concurrently. AIOProducer has a two-step result: the first await enqueues, while the returned future resolves after delivery. Version 2.15.1 now uses asyncio.get_running_loop() and fails when no loop exists instead of creating one. ShareConsumer remains synchronous, and explicit mode refuses the next poll until every record in the previous batch has an acknowledgement.
Patterns
Send records and observe broker delivery produce-with-delivery-report
from confluent_kafka import Producer
producer = Producer({"bootstrap.servers": "broker1:9092,broker2:9092"})
def delivered(error, message):
if error is not None:
print("failed:", error)
return
print(message.topic(), message.partition(), message.offset())
for event in events:
producer.poll(0)
producer.produce(
"orders",
key=str(event.id).encode(),
value=event.to_json().encode(),
on_delivery=delivered,
)
remaining = producer.flush(15)
if remaining:
raise RuntimeError(f"{remaining} records were not delivered")`produce()` confirms only local enqueueing. Delivery callbacks run from `poll()` or `flush()`, and `flush(15)` returns the records left after 15 seconds.
Retry after the producer queue fills recover-from-full-queue
while True:
try:
producer.produce("orders", value=payload, on_delivery=delivered)
break
except BufferError:
producer.poll(0.5)
producer.poll(0)`BufferError` means the local producer queue has reached its configured limit. Polling serves completed deliveries and may free a slot for the retry.
Poll a consumer and preserve error handling consume-records
from confluent_kafka import Consumer, KafkaError, KafkaException
consumer = Consumer({
"bootstrap.servers": "broker1:9092",
"group.id": "order-indexer",
"auto.offset.reset": "earliest",
})
consumer.subscribe(["orders"])
try:
while running:
message = consumer.poll(1.0)
if message is None:
continue
if message.error():
if message.error().code() == KafkaError._PARTITION_EOF:
continue
raise KafkaException(message.error())
index_order(message.value())
finally:
consumer.close()`poll(1.0)` can return `None` or a Message whose `error()` is set. `close()` leaves the consumer group and completes its local cleanup.
Commit only after the side effect succeeds commit-after-processing
consumer = Consumer({
"bootstrap.servers": "broker1:9092",
"group.id": "order-indexer",
"enable.auto.commit": False,
})
consumer.subscribe(["orders"])
message = consumer.poll(1.0)
if message is not None and not message.error():
write_to_database(message.value())
consumer.commit(message=message, asynchronous=False)Disabling auto commit keeps the offset behind the database write. A synchronous commit waits for the broker, so commit batches when one round trip per record is too costly.
Authenticate to Confluent Cloud configure-confluent-cloud
import os
kafka_config = {
"bootstrap.servers": os.environ["KAFKA_BOOTSTRAP_SERVERS"],
"security.protocol": "SASL_SSL",
"sasl.mechanisms": "PLAIN",
"sasl.username": os.environ["KAFKA_API_KEY"],
"sasl.password": os.environ["KAFKA_API_SECRET"],
"client.id": "order-api",
}Kafka API credentials belong in this client configuration. Schema Registry has a different URL and key pair, so do not reuse these values in `basic.auth.user.info`.
Encode Avro with Schema Registry serialize-avro
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 MessageField, SerializationContext
registry = SchemaRegistryClient({
"url": schema_registry_url,
"basic.auth.user.info": schema_registry_credentials,
})
serializer = AvroSerializer(
registry,
schema_str,
lambda order, ctx: {"id": order.id, "total": order.total},
)
producer = Producer(kafka_config)
producer.produce(
"orders",
value=serializer(order, SerializationContext("orders", MessageField.VALUE)),
)
producer.flush()Install `confluent-kafka[avro,schemaregistry]` for this path. The serialization context supplies the topic and key-or-value position used by subject naming.
Commit output and consumed offsets together process-transactionally
producer = Producer({
"bootstrap.servers": brokers,
"transactional.id": "order-normalizer-1",
})
producer.init_transactions()
producer.begin_transaction()
try:
for message in batch:
producer.produce("orders-clean", value=normalize(message.value()))
producer.send_offsets_to_transaction(
consumer.position(consumer.assignment()),
consumer.consumer_group_metadata(),
)
producer.commit_transaction()
except Exception:
producer.abort_transaction()
raiseTurn off consumer auto commit before this flow. Each active producer needs its own stable `transactional.id`, and downstream consumers need `isolation.level=read_committed` to hide aborted output.
Await enqueueing and delivery separately produce-with-asyncio
import asyncio
from confluent_kafka.aio import AIOProducer
async def send_order(payload):
producer = AIOProducer({"bootstrap.servers": "broker1:9092"})
try:
delivery = await producer.produce("orders", value=payload)
message = await delivery
return message.partition(), message.offset()
finally:
await producer.flush()
await producer.close()
asyncio.run(send_order(b"order-42"))The first await puts the record into the producer path and returns a future. The second await receives the delivered Message or raises its delivery error.
Run a consumer without blocking the event loop consume-with-asyncio
from confluent_kafka.aio import AIOConsumer
async def consume_orders():
consumer = AIOConsumer({
"bootstrap.servers": "broker1:9092",
"group.id": "async-worker",
"enable.auto.commit": "false",
})
await consumer.subscribe(["orders"])
try:
while running:
message = await consumer.poll(1.0)
if message is None:
continue
if message.error():
raise RuntimeError(message.error())
await handle_order(message.value())
await consumer.commit(message=message)
finally:
await consumer.close()AIOConsumer operations are awaitable and callbacks run on the event loop. Version 2.15.1 requires an already running loop rather than creating one implicitly.
Check every topic-creation result create-topics
from confluent_kafka.admin import AdminClient, NewTopic
admin = AdminClient({"bootstrap.servers": "broker1:9092"})
requests = [
NewTopic("orders", num_partitions=6, replication_factor=3),
NewTopic("order-events", num_partitions=6, replication_factor=3),
]
for topic, future in admin.create_topics(requests, request_timeout=15).items():
try:
future.result()
print("created", topic)
except Exception as error:
print("failed", topic, error)`create_topics()` returns one future per topic. Calling each future's `result()` is where authorization, validation, and broker errors surface.
Read again from a known offset replay-partition
from confluent_kafka import Consumer, KafkaException, TopicPartition
consumer = Consumer({
"bootstrap.servers": "broker1:9092",
"group.id": "repair-job",
})
start = TopicPartition("orders", 0, 15_000)
consumer.assign([start])
try:
low, high = consumer.get_watermark_offsets(start)
if not low <= start.offset < high:
raise ValueError(f"offset must be between {low} and {high - 1}")
next_offset = start.offset
while next_offset < high:
message = consumer.poll(1.0)
if message is None:
continue
if message.error():
raise KafkaException(message.error())
repair(message.value())
next_offset = message.offset() + 1
finally:
consumer.close()`assign()` directly selects partitions and suits a repair job that should bypass group assignment. Validate the requested offset against the returned low and high watermarks before a long replay.
Acknowledge preview ShareConsumer records acknowledge-shared-records
from confluent_kafka import AcknowledgeType, ShareConsumer
consumer = ShareConsumer({
"bootstrap.servers": "broker1:9092",
"group.id": "shared-workers",
"share.acknowledgement.mode": "explicit",
})
consumer.subscribe(["jobs"])
with consumer:
for message in consumer.poll(timeout=1.0):
if message.error():
continue
try:
run_job(message.value())
consumer.acknowledge(message, AcknowledgeType.ACCEPT)
except TemporaryFailure:
consumer.acknowledge(message, AcknowledgeType.RELEASE)
results = consumer.commit_sync(timeout=10.0)ShareConsumer is preview-only in 2.15.1 and requires Kafka 4.2.0 share groups. Explicit mode requires an acknowledgement for every record before the next poll.
Alternatives
| Package | Registry | Pick it when |
|---|---|---|
| aiokafka | PyPI | Pick it when an asyncio-first Python API matters more than using librdkafka or Confluent's Schema Registry clients. |
| kafka-python | PyPI | Pick it when the deployment must stay pure Python and its Kafka feature coverage is enough for the service. |
| faust-streaming | PyPI | Pick it when the code should express agents, tables, and stream transformations above Kafka consumption. |
| fastavro | PyPI | Pick it for Avro files or schemaless records when neither Kafka transport nor Schema Registry belongs in the job. |
More data guides
numpy · pandas · fsspec · pyarrow · lxml · sqlalchemy · 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.