Python Confluent Kafka Producer Consumer and Schema Registry

When you build a Kafka pipeline in Python, the combination of the confluent-kafka client and Schema Registry solves a problem that raw Kafka does not: agreeing on the structure of the messages. Without a schema, producers and consumers must share a serialization format by convention, and any change to the message shape can break consumers silently. This article walks through a working python confluent kafka producer consumer and schema registry setup, using Avro as the serialization format, and covers the operational details that matter when you move from a proof of concept to a production stream.

Why Schema Registry Belongs in a Kafka Pipeline

Kafka itself is a byte-array transport. A producer can send any payload, and a consumer receives bytes with no built-in interpretation. That flexibility becomes a liability as soon as more than one service produces or consumes a topic. A producer might add a field, rename one, or change a type, and every consumer must be updated in lockstep or risk deserialization failures.

Schema Registry centralizes the contract. Producers register a schema for each topic, and consumers fetch that schema to deserialize messages. The registry enforces compatibility rules, so you can evolve the schema without breaking existing consumers. In the Python ecosystem, the confluent-kafka library provides Serializer and Deserializer classes that integrate directly with the registry, so you rarely handle raw bytes yourself.

Avro is the most common choice with Schema Registry because it has a compact binary encoding and a well-defined schema evolution model. The same pattern works with Protobuf or JSON Schema, but the examples here use Avro because it is the default for most Kafka deployments.

Setting Up the Client and Registry Connections

The confluent-kafka package is the official Python client for Kafka, and it includes the schema registry modules. Install it with pip:

pip install confluent-kafka[avro]

The [avro] extra pulls in fastavro and the schema registry client. You need two separate connections: one to the Kafka brokers and one to the Schema Registry service. Both are configured through dictionaries passed to the producer or consumer constructors.

from confluent_kafka import Producer, Consumer
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer, AvroDeserializer

kafka_config = {
    'bootstrap.servers': 'broker1:9092,broker2:9092',
    'security.protocol': 'SASL_SSL',
    'sasl.mechanism': 'PLAIN',
    'sasl.username': 'user',
    'sasl.password': 'password',
}

schema_registry_conf = {
    'url': 'https://schema-registry.example.com',
    'basic.auth.user.info': 'key:secret',
}

schema_registry_client = SchemaRegistryClient(schema_registry_conf)

The SchemaRegistryClient handles the HTTP calls to the registry, caching schemas locally by ID. You do not need to manage schema IDs manually; the serializer and deserializer use the client to fetch and cache them.

Producing Avro Messages with the Confluent Producer

To produce Avro messages, you create an AvroSerializer with a schema string and the registry client. The serializer converts a Python dictionary into Avro bytes. You then pass that serializer to the Producer via the value.serializer configuration.

value_schema = """
{
  "type": "record",
  "name": "Order",
  "namespace": "com.example.orders",
  "fields": [
    {"name": "order_id", "type": "string"},
    {"name": "customer_id", "type": "string"},
    {"name": "amount", "type": "double"},
    {"name": "created_at", "type": "long"}
  ]
}
"""

avro_serializer = AvroSerializer(schema_registry_client, value_schema)

producer = Producer({**kafka_config, 'value.serializer': avro_serializer})

def delivery_report(err, msg):
    if err is not None:
        print(f'Delivery failed: {err}')
    else:
        print(f'Message delivered to {msg.topic()} [{msg.partition()}]')

order = {
    'order_id': 'ORD-1001',
    'customer_id': 'CUST-42',
    'amount': 199.99,
    'created_at': 1700000000000
}

producer.produce('orders', value=order, callback=delivery_report)
producer.flush()

The produce method takes the topic name and a Python dictionary. The serializer encodes it to Avro, registers the schema if it is new, and prepends the schema ID to the message bytes. The callback runs on delivery confirmation, which is the correct place to check for errors rather than assuming success after produce returns.

flush() blocks until all pending messages are delivered. In a long-running service, you would call poll() periodically to serve delivery callbacks and avoid unbounded memory growth.

Consuming Avro Messages with the Confluent Consumer

Consuming requires an AvroDeserializer configured with the same registry client. The deserializer reads the schema ID from the message bytes, fetches the schema if it is not cached, and converts the Avro payload back into a Python dictionary.

avro_deserializer = AvroDeserializer(schema_registry_client, value_schema)

consumer = Consumer({**kafka_config, 'value.deserializer': avro_deserializer})
consumer.subscribe(['orders'])

while True:
    msg = consumer.poll(timeout=1.0)
    if msg is None:
        continue
    if msg.error():
        print(f'Consumer error: {msg.error()}')
        continue

    order = msg.value()
    print(f'Received order {order["order_id"]} for {order["customer_id"]} amount {order["amount"]}')

The consumer fetches the schema automatically from the registry. If you do not pass a schema to the deserializer, it will still work because the schema ID is embedded in the message, but passing it avoids a registry round-trip for the first message of each schema ID.

Note that the consumer must be assigned to a group to manage offsets. The default group.id is not set, so you must provide one in kafka_config. Offsets are committed automatically by default, but for production you may want to control commit timing.

Schema Evolution and Compatibility Rules

Schema Registry does more than store schemas; it enforces a compatibility level per subject. A subject is the combination of topic and key/value namespace, typically named topic-name-value. When you produce a message with a new schema, the registry checks it against the latest schema for that subject.

The common levels are:

• BACKWARD: consumers using the new schema can read data written with the old schema. This allows adding fields with defaults or removing fields.
• FORWARD: consumers using the old schema can read data written with the new schema. This allows removing fields or adding optional fields.
• FULL: both backward and forward compatibility.
• NONE: no checks.

For Avro, adding a field with a default is backward compatible. Removing a field is forward compatible. Changing a type without a logical conversion is usually incompatible.

When a producer sends a message with a schema that violates the compatibility level, the registry returns an error, and the producer fails with a SchemaRegistryError. The serializer does not silently send bytes; it raises during produce(). This is a key operational difference from plain Kafka, where malformed data would be sent and consumers would fail later.

To evolve a schema, update the schema string in your producer code and restart. The registry will accept it if the compatibility rules pass. Consumers with the older schema will still be able to deserialize new messages as long as the compatibility level is BACKWARD or FULL.

Error Handling and Delivery Guarantees

Kafka's delivery semantics depend on producer configuration, not on the serializer. The confluent-kafka producer exposes delivery.timeout.ms, acks, and enable.idempotence to control durability and ordering.

For exactly-once semantics, you would need to use Kafka transactions, which the Python client supports through init_transactions() and begin_transaction(). However, most applications do not need that level and rely on at-least-once with idempotent producers.

Common producer errors include:

• MessageSizeTooLargeError when a message exceeds message.max.bytes.
• Local: Queue full when the producer buffer is full and queue.buffering.max.messages is reached.
• SchemaRegistryError when schema registration fails due to compatibility or network issues.

On the consumer side, the most important error is _ALL_BROKERS_DOWN or _TRANSPORT errors, which indicate connectivity problems. Deserialization errors appear as msg.value() returning None or raising an exception, depending on the deserializer's from_dict behavior. You should always check msg.error() before accessing msg.value().

Delivery callbacks are the only reliable way to know if a message was accepted by the broker. Do not rely on produce() returning without raising; it only means the message was buffered locally.

Production Configuration and Performance Considerations

Several settings have a direct impact on throughput and reliability when using the Confluent Python client.

Producer Batching and Buffering

The producer batches messages per partition. linger.ms controls how long the producer waits to accumulate more messages before sending a batch. A higher value increases throughput at the cost of latency. batch.num.messages sets the maximum number of messages per batch.

In a Python process, the producer runs background threads for network I/O, so produce() is non-blocking. You must call poll() or flush() regularly to trigger delivery callbacks and free up buffer space. A common pattern is to call producer.poll(0) after each produce() in a tight loop.

Consumer Fetch and Commit Behavior

Consumers fetch messages in batches controlled by fetch.min.bytes and fetch.max.wait.ms. For low latency, set fetch.min.bytes to 1 and fetch.max.wait.ms to a low value. For higher throughput, allow larger batches.

Offset commits can be automatic or manual. Automatic commits are convenient but can cause duplicate processing if a consumer crashes between processing and the next auto-commit. Manual commits with commit() after processing give you at-least-once semantics with better control. Use enable.auto.commit=false and commit after your business logic succeeds.

Schema Registry Caching

Both serializer and deserializer cache schemas by ID. The first message for a new schema triggers a registry lookup, which adds latency. In a high-throughput environment, that one-time cost is negligible. However, if your registry is slow or unreachable, producers and consumers will fail. Always configure timeouts for the registry client, such as request.timeout.ms and max.retries.

Security and Authentication

Production Kafka clusters typically require SASL/SSL. The confluent-kafka client supports SASL_SSL with PLAIN, SCRAM, or OAUTHBEARER. Schema Registry also requires authentication, often with basic auth or mTLS. Store credentials in environment variables or a secrets manager, never in source code.

Idempotent Producer

Set enable.idempotence=true to prevent duplicate messages due to retries. This ensures the broker deduplicates messages using a sequence number, but it requires acks=all and a single in-flight request per partition. This is a safe default for most applications.

Handling Key Serialization

So far, the examples only serialize the value. In Kafka, the key is optional but important for partitioning. If you want messages with the same key to go to the same partition, you must serialize the key as well. The same AvroSerializer can be used for keys, but you typically use a simpler schema, such as a string or a record with a single field.

key_schema = """
{
  "type": "record",
  "name": "OrderKey",
  "namespace": "com.example.orders",
  "fields": [
    {"name": "order_id", "type": "string"}
  ]
}
"""

key_serializer = AvroSerializer(schema_registry_client, key_schema)
producer = Producer({**kafka_config, 'key.serializer': key_serializer, 'value.serializer': avro_serializer})

producer.produce('orders', key={'order_id': 'ORD-1001'}, value=order)

The consumer must also deserialize the key with a matching key.deserializer. If you do not set a key serializer, the key is sent as None and Kafka uses a round-robin partitioner.

When to Use Schema Registry vs. Plain Serialization

Schema Registry adds operational complexity: you must run and monitor a service, manage compatibility levels, and handle schema registration latency. For a single producer and consumer in a controlled environment, you might get away with a simple JSON or pickle serialization. But as soon as multiple teams produce or consume a topic, or you need to evolve the data contract without downtime, Schema Registry becomes the safer choice.

The decision comes down to whether the cost of coordinating schema changes manually exceeds the cost of running the registry. In most event-driven architectures, the registry pays for itself by preventing silent breakage. The Python confluent-kafka client makes the integration straightforward, so the main effort is designing your schemas and compatibility rules up front.