Confluent Python Client for Apache Kafka

Confluent's Python Client for Apache Kafka<sup>TM</sup>

confluent-kafka-python provides a high-level Producer, Consumer and AdminClient compatible with all Apache Kafka™ brokers >= v0.8, Confluent Cloud and Confluent Platform.

Recommended for Production: While this client works with any Kafka deployment, it's optimized for and fully supported with Confluent Cloud (fully managed) and Confluent Platform (self-managed), which provide enterprise-grade security, monitoring, and support.

Why Choose Confluent's Python Client?

Unlike the basic Apache Kafka Python client, confluent-kafka-python provides:

Production-Ready Performance: Built on librdkafka (C library) for maximum throughput and minimal latency, significantly outperforming pure Python implementations.
Enterprise Features: Schema Registry integration, transactions, exactly-once semantics, and advanced serialization support out of the box.
AsyncIO Support: Native async/await support for modern Python applications - not available in the Apache Kafka client.
Comprehensive Serialization: Built-in Avro, Protobuf, and JSON Schema support with automatic schema evolution handling.
Professional Support: Backed by Confluent's engineering team with enterprise SLAs and 24/7 support options.
Active Development: Continuously updated with the latest Kafka features and performance optimizations.
Battle-Tested: Used by thousands of organizations in production, from startups to Fortune 500 companies.

Performance Note: The Apache Kafka Python client (kafka-python) is a pure Python implementation that, while functional, has significant performance limitations for high-throughput production use cases. confluent-kafka-python leverages the same high-performance C library (librdkafka) used by Confluent's other clients, providing enterprise-grade performance and reliability.

Key Features

High Performance & Reliability: Built on librdkafka, the battle-tested C client for Apache Kafka, ensuring maximum throughput, low latency, and stability. The client is supported by Confluent and is trusted in mission-critical production environments.
Comprehensive Kafka Support: Full support for the Kafka protocol, transactions, and administration APIs.
AsyncIO Producer: A fully asynchronous producer (AIOProducer) for seamless integration with modern Python applications using asyncio.
Seamless Schema Registry Integration: Synchronous and asynchronous clients for Confluent Schema Registry to handle schema management and serialization (Avro, Protobuf, JSON Schema).
Improved Error Handling: Detailed, context-aware error messages and exceptions to speed up debugging and troubleshooting.
[Confluent Cloud] Automatic Zone Detection: Producers automatically connect to brokers in the same availability zone, reducing latency and data transfer costs without requiring manual configuration.
[Confluent Cloud] Simplified Configuration Profiles: Pre-defined configuration profiles optimized for common use cases like high throughput or low latency, simplifying client setup.
Enterprise Support: Backed by Confluent's expert support team with SLAs and 24/7 assistance for production deployments.

Usage

For a step-by-step guide on using the client, see Getting Started with Apache Kafka and Python.

Choosing Your Kafka Deployment

Confluent Cloud - Fully managed service with automatic scaling, security, and monitoring. Best for teams wanting to focus on applications rather than infrastructure.
Confluent Platform - Self-managed deployment with enterprise features, support, and tooling. Ideal for on-premises or hybrid cloud requirements.
Apache Kafka - Open source deployment. Requires manual setup, monitoring, and maintenance.

Additional examples can be found in the examples directory or the confluentinc/examples GitHub repo, which include demonstrations of:

Exactly once data processing using the transactional API.
Integration with asyncio.
(De)serializing Protobuf, JSON, and Avro data with Confluent Schema Registry integration.
Confluent Cloud configuration.

Also see the Python client docs and the API reference.

Finally, the tests are useful as a reference for example usage.

AsyncIO Producer

Use the AsyncIO Producer inside async applications to avoid blocking the event loop.

import asyncio
from confluent_kafka.aio import AIOProducer

async def main():
    p = AIOProducer({"bootstrap.servers": "mybroker"})
    try:
        # produce() returns a Future; first await the coroutine to get the Future,
        # then await the Future to get the delivered Message.
        delivery_future = await p.produce("mytopic", value=b"hello")
        delivered_msg = await delivery_future
        # Optionally flush any remaining buffered messages before shutdown
        await p.flush()
    finally:
        await p.close()

asyncio.run(main())

Notes:

Batched async produce buffers messages; delivery callbacks, stats, errors, and logger run on the event loop.
Per-message headers are not supported in the batched async path. If headers are required, use the synchronous Producer.produce(...) (you can offload to a thread in async apps).

For a more detailed example that includes both an async producer and consumer, see examples/asyncio_example.py.

Architecture: For implementation details and component architecture, see the AIOProducer Architecture Overview.

When to use AsyncIO vs synchronous Producer

Use AsyncIO Producer when your code runs under an event loop (FastAPI/Starlette, aiohttp, Sanic, asyncio workers) and must not block.
Use synchronous Producer for scripts, batch jobs, and highest-throughput pipelines where you control threads/processes and can call poll()/flush() directly.
In async servers, prefer AsyncIO Producer; if you need headers, call sync produce() via run_in_executor for that path.

AsyncIO with Schema Registry

The AsyncIO producer and consumer integrate seamlessly with async Schema Registry serializers. See the Schema Registry Integration section below for full details.

Basic Producer example

from confluent_kafka import Producer

p = Producer({'bootstrap.servers': 'mybroker1,mybroker2'})

def delivery_report(err, msg):
    """ Called once for each message produced to indicate delivery result.
        Triggered by poll() or flush()."""
    if err is not None:
        print('Message delivery failed: {}'.format(err))
    else:
        print('Message delivered to {} [{}]'.format(msg.topic(), msg.partition()))

for data in some_data_source:
    # Trigger any available delivery report callbacks from previous produce() calls
    p.poll(0)

    # Asynchronously produce a message. The delivery report callback will
    # be triggered from the call to poll() above, or flush() below, when the
    # message has been successfully delivered or failed permanently.
    p.produce('mytopic', data.encode('utf-8'), callback=delivery_report)

# Wait for any outstanding messages to be delivered and delivery report
# callbacks to be triggered.
p.flush()

For a discussion on the poll based producer API, refer to the Integrating Apache Kafka With Python Asyncio Web Applications blog post.

Schema Registry Integration

This client provides full integration with Schema Registry for schema management and message serialization, and is compatible with both Confluent Platform and Confluent Cloud. Both synchronous and asynchronous clients are available.

Learn more

Getting Started with Apache Kafka and Python – step-by-step course with videos and labs.
Schema Registry Basics – short tutorial explaining subjects, compatibility, and serialization patterns.

Synchronous Client & Serializers

Use the synchronous SchemaRegistryClient with the standard Producer and Consumer.

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 StringSerializer, SerializationContext, MessageField

# Configure Schema Registry Client
schema_registry_conf = {'url': 'http://localhost:8081'}  # Confluent Platform
# For Confluent Cloud, add: 'basic.auth.user.info': '<sr-api-key>:<sr-api-secret>'
# See: https://docs.confluent.io/cloud/curren

confluent-kafka

Description