How Kafka Handles Schema Evolution with Avro Schemas: A Complete Implementation Guide

Kafka delegates Avro schema storage to the Confluent Schema Registry, which assigns unique numeric IDs to schema versions and enforces compatibility rules, allowing producers to embed these IDs in message payloads while consumers fetch the correct schema version on demand.

The DataTalksClub/data-engineering-zoomcamp repository demonstrates production-ready patterns for implementing Kafka schema evolution with Avro in Python streaming pipelines. This architecture separates data from metadata by externalizing schema management, enabling independent deployment of producers and consumers without data corruption.

How the Confluent Schema Registry Enables Schema Evolution

Kafka itself does not store schema information. Instead, the Confluent Schema Registry acts as a centralized service that manages Avro schemas, assigns unique version IDs, and validates compatibility between changes.

When a producer serializes a record using AvroSerializer, it automatically registers the schema with the registry and receives a schema ID that gets written into the Kafka message payload. When a consumer processes the message, AvroDeserializer reads this ID, fetches the corresponding schema from the registry, and deserializes the payload using that specific version.

This decoupling allows you to evolve schemas safely as long as the new version respects the configured compatibility rules:

• BACKWARD – New consumers can read data written with the old schema
• FORWARD – Old consumers can read data written with the new schema (requires default values for added fields)
• FULL – Both directions are allowed simultaneously

Producer Implementation with AvroSerializer

In 07-streaming/extras/python/avro_example/producer.py, the producer creates a SchemaRegistryClient, loads the Avro schema files, and initializes serializers that automatically handle schema registration and ID embedding.

# 07-streaming/extras/python/avro_example/producer.py (lines 27-34)

schema_registry_props = {'url': props['schema_registry.url']}
schema_registry_client = SchemaRegistryClient(schema_registry_props)
self.key_serializer = AvroSerializer(schema_registry_client, key_schema_str, ride_record_key_to_dict)
self.value_serializer = AvroSerializer(schema_registry_client, value_schema_str, ride_record_to_dict)

The AvroSerializer handles the handshake with the registry. If the schema is new, it registers automatically and caches the resulting ID for subsequent messages. This ID travels with every Kafka record, ensuring the consumer knows exactly which schema version to use for deserialization.

Consumer Implementation with AvroDeserializer

The consumer setup in 07-streaming/extras/python/avro_example/consumer.py mirrors the producer pattern but focuses on retrieving schemas using the embedded ID.

# 07-streaming/extras/python/avro_example/consumer.py (lines 22-28)

self.avro_key_deserializer = AvroDeserializer(
      schema_registry_client=schema_registry_client,
      schema_str=key_schema_str,
      from_dict=dict_to_ride_record_key)
self.avro_value_deserializer = AvroDeserializer(
      schema_registry_client=schema_registry_client,
      schema_str=value_schema_str,
      from_dict=dict_to_ride_record)

When AvroDeserializer encounters a message, it extracts the schema ID from the payload, fetches the corresponding definition from the registry, and applies any default values for missing fields. This allows older consumers to read data produced with newer schema versions (forward compatibility) or newer consumers to process old data (backward compatibility).

Step-by-Step Schema Evolution Example

To evolve a schema while maintaining compatibility, you modify the Avro schema definition, ensure new fields have default values, and deploy the new producer configuration.

First, create a new schema file with an optional field and default value:

# resources/schemas/taxi_ride_value_v2.avsc

{
  "type": "record",
  "name": "Ride",
  "fields": [
    {"name": "vendor_id", "type": "int"},
    {"name": "pickup_datetime", "type": "string"},
    {"name": "dropoff_datetime", "type": "string"},
    {"name": "passenger_count", "type": ["null", "int"], "default": null},
    {"name": "payment_type", "type": ["null", "string"], "default": null}
  ]
}

Next, configure the producer to use the new schema path:

config = {
    'bootstrap.servers': BOOTSTRAP_SERVERS,
    'schema_registry.url': SCHEMA_REGISTRY_URL,
    'schema.key': RIDE_KEY_SCHEMA_PATH,
    'schema.value': 'resources/schemas/taxi_ride_value_v2.avsc'
}
producer = RideAvroProducer(props=config)
producer.publish(topic=KAFKA_TOPIC, records=ride_records)

Consumers can remain unchanged and will continue using their cached schema version. If upgraded, they will automatically fetch the new schema version and receive None for the payment_type field when deserializing older records that lack this field.

Configuration and Compatibility Management

The settings.py file in the repository handles the connection parameters for both Kafka and the Schema Registry. Compatibility rules are typically configured via the Schema Registry REST API or web interface, not in client code.

Critical compatibility requirements for schema evolution:

• Adding fields – Must include a default value or be a union with null to maintain forward compatibility
• Removing fields – Must have been optional (union with null) in the previous version to maintain backward compatibility
• Changing types – Generally requires a new subject in the registry rather than evolution

Summary

• Kafka does not store schemas internally; the Confluent Schema Registry manages Avro schema versions and assigns numeric IDs.
• Producers use AvroSerializer to register schemas and embed schema IDs in message payloads via 07-streaming/extras/python/avro_example/producer.py.
• Consumers use AvroDeserializer to read schema IDs and fetch the correct version from the registry, as shown in 07-streaming/extras/python/avro_example/consumer.py.
• Schema evolution requires compatibility modes (BACKWARD, FORWARD, FULL) and default values for new fields to ensure old consumers can read new data.
• The DataTalksClub/data-engineering-zoomcamp implementation stores schema files in resources/schemas/ and references them through configuration in settings.py.

Frequently Asked Questions

Does Kafka store Avro schemas internally?

No. Kafka stores only the binary message payload and the schema ID. The actual Avro schema definitions reside in the Confluent Schema Registry, which operates as a separate service. This separation allows schemas to evolve without requiring topic migration or consumer group rebalancing.

What happens if I add a required field to an Avro schema without a default value?

Adding a required field breaks forward compatibility. Old consumers will fail to deserialize new messages because they expect the field to be present. To safely evolve schemas, always add new fields as optional (union with null) or provide explicit default values in the Avro schema definition.

How does the consumer know which schema version to use for each message?

The producer embeds a numeric schema ID in each Kafka message payload. When the consumer deserializes the message, AvroDeserializer extracts this ID and queries the Schema Registry to fetch the corresponding schema version. This happens transparently without requiring consumer configuration changes when schemas evolve compatibly.

Where are the Schema Registry compatibility settings configured?

Compatibility rules are configured at the Schema Registry level, either through the REST API (/config endpoints) or the Confluent Control Center web interface. Client applications in producer.py and consumer.py do not set these rules; they simply respect the enforcement performed by the registry when attempting to register new schema versions.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}