Schemas, Schema Registries and Kafka - Part Three

24 October 2024 // Martin Holt

In part one and part two we looked at Schemas and the Schema Registry. In part three we will take a look at schema evolution in Kafka with the help of an example.

What is Schema Evolution?

Schema evolution allows a schema to change over time with few or no consequences for the processes that rely on that schema. This blog is focuses on Kafka with a Producer - Consumer scenario in which some information needs to be changed, added or removed in an existing schema. Any change risks violating the existing contract between Producer and Consumer which may cause the Consumer to fail, possibly in a catastrophic way.

Schema Versions

At a given point in time there is no guarantee that a Consumer and Producer are using the exact same schema. If we introduce the concept of a schema version we can say that the Producer is using one version of a schema and the Consumer another or the same version. For these two components to continue working smoothly independently of each other we must offer some compatibility guarantees between schea versions. These guarantees can be provided by the Schema Registry.

The Use Case

Continuing our weather reporting scenario imagine that we have devices publishing weather reports from different locations. The weather reports are consumed by a single central service. Again we will consider schemas in Avro, Protobuf and JSON schema format. As before a Schema Registry will be used to enforce schema compliance - if the Schema Registry contains the schema then the data can be written or read from a topic.

In the scenario we have been producing weather reports for a while but have identified some fields that we aren’t satisfied with. This has led to a new proposal for the weather report schemas - the schemas will be versioned.

Unfortunately for us the weather reporting devices need a manual update. This requires a technician to travel to each location and so it is guaranteed that at any one point in time Producers could be using either the new or old schema version.

The good news though is that we have a single subscriber service and that they are ready to use the new version of the schemas.

How can we implement this?

Compatibility Guaranteees

In the scenario our Consumer will need to handle both versions of the schema. Let’s start by assigning these versions an identity, “alpha” for the previous version and “beta” or the latest version.

The scenario is again using the Apicurio Registry and it will be the registry that ensures schema compatibility. Compatibility guarantees are not standardised so once again you will need to get to know your chosen registry and what it provides.

The Apicurio Registry provides a compatibility rule which can be used to perform a compatibility check when a schema is updated. For our purposes we will enforce BACKWARD compatibility which is defined as:

Clients using the new artifact can read data written using the most recently added artifact.

Note that this scenario assumes that the Consumer has been prepared in advance of the “beta” version being used by a Producer. This is quite an assumption and will require coordination between the affected systems. Had we a scenario with multiple Consumers or could not identify all Consumers we may have chosen FULL compatibility.

A full investigation of all the compatibility guarantees is out of scope for this blog but a simple rule of thumb is the higher the level of compatibility the more complex things become in the schema (nearly every field becomes optional) and the more defensive code you will need to write (for Java developers you will be using Optional.ofNullable(...) a lot).

Setting Up The Registry

As before the examples used in this repository can be found in this Github repository.

The working example used to illustrate this use case will only look at the Schema Registry component and how it handles the compatibility guarantees.

To illustrate what is happening the working example let us set up a seperate Schema Registry without any Consumers and Producers. Use Docker to start a container for the schema registry:

> docker run -d -p 8082:8080 quay.io/apicurio/apicurio-registry-mem:2.6.4.Final

This example will use the APIs to access the Apicurio registry but there is also a UI available at http://localhost:8082/ui if preferred.

Enforcing Compatibility

A compatibility rule can be configured for each schema. However to make things easier for us let’s use a Global Rule to enforce backward compatibility for all schemas:

> curl -X POST -H "content-type: application/json" --data @./apicurio/compatibility_rule.json http://localhost:8082/apis/registry/v2/admin/rules

These rules can then be inspected in the UI if so desired.

Publish the Alpha Versions

Next step is to publish the “alpha” versions of the schemas which can be found in the /schemas/alpha/ folder. To do this we use the Apicurio REST Api (or UI if you prefer):

> curl -X POST -H "X-Registry-ArtifactType=JSON" \
               -H "X-Registry-ArtifactId: weather-json-schema" \
               -H "X-Registry-Version: alpha" \
               --data-binary @./schemas/alpha/weather-schema-v1.json \
               "http://localhost:8082/apis/registry/v2/groups/weather/artifacts"

> curl -X POST -H "X-Registry-ArtifactType=AVRO" \
               -H "X-Registry-ArtifactId: weather-avro-schema" \
               -H "X-Registry-Version: alpha" \
               --data-binary @./schemas/alpha/weather-schema.avsc \
               "http://localhost:8082/apis/registry/v2/groups/weather/artifacts"

> curl -X POST -H "X-Registry-ArtifactType=PROTOBUF" \
               -H "X-Registry-ArtifactId: weather-proto-schema" \
               -H "X-Registry-Version: alpha" \
               --data-binary @./schemas/alpha/weather-report.proto \
               "http://localhost:8082/apis/registry/v2/groups/weather/artifacts"

What have we done? We have chosen to add three schemas to the group weather. We have explicitly given these artifacts an identify and a version of “alpha”.

Note that the version is not a part of the schema itself - it is not the schema that is being versioned it is the schema registry representation of the schema. Note also that the schema version is simply a name. To the schema registry the eldest version is the version that was added first. If the schemas were added with “beta” first the registry would consider the “beta” version to be the eldest.

One tip to make debugging easier is to add the version into the schema, using some form of metadata, and to log that when serializing / deserializing. This will help when debugging the Consumers and Producers.

Breaking Backward Compatibility

Now that we have a standalone schema registry that enforces BACKWARD compatiblilty and the “alpha” versions of the schemas loaded let’s see how enforcement works.

Our weather reporting schemas all contain some form of optional observations fields. It might seem odd to have a weather report with no observations, so let’s make these fields mandatory. The changed schemas can be found in the /schemas/non-compatible directory and can be added as “beta” versions of the schema through the UI or through the APIs like this:

> curl -X PUT -H "X-Registry-Version: beta" \
               --data-binary @./schemas/non-compatible/weather-schema-non-backward.json \
               "http://localhost:8082/apis/registry/v2/groups/weather/artifacts/weather-json-schema"

> curl -X PUT -H "X-Registry-Version: beta" \
               --data-binary @./schemas/non-compatible/weather-schema-non-backward.avsc \
               "http://localhost:8082/apis/registry/v2/groups/weather/artifacts/weather-avro-schema"

> curl -X PUT -H "X-Registry-Version: beta" \
               --data-binary @./schemas/non-compatible/weather-report-non-backward.proto \
               "http://localhost:8082/apis/registry/v2/groups/weather/artifacts/weather-proto-schema"

Making an optional field mandatory breaks backward compatibility - a message produced using the “alpha” version with no observations cannot be consumed using the “beta” version of the schema. Trying to make the observations field mandatory will give you an error message when uploading the new schema version - as illustrated by the UI here:

Compatibility enforcement by the Schema Registry will give you confidence when evolving your schemas. Incompatible versions of the schema cannot be added, and therefore cannot be used by the Consumers or Producer.

Note that there is nothing here that stops you from testing your schema evolution prior to taking it to the production environment, and even automating it as part of your pipeline (just remember to ensure your test schema registry is loaded in the same order as the production registry).

Evolving The Schemas

Let’s try some changes that should be compatible. After some consideration the subjective visibility measurement is to be retired and replaced with a more precise measurement visibilityDistance. We also noticed that the observations value precipitationTotal24hh is misspelled and we would like to fix that. Updated schemas can be found in the /schemas/beta/ folder.

For the Avro case we can simply remove the optional visibility field, and add a new optional visibilityDistance field. We can also rename the precipitationTotal24hh field and use an alias with the old name.

{
  "name": "observations",
  ...
    {
      "name": "precipitationTotal24h",
      "type": [
        "double",
        "null"
      ],
      "aliases": ["precipitationTotal24hh"],
      "doc": "Total rain in preceding 24 hours"
    },
    ...
    {
      "name": "visibilityDistance",
      "type": [
        "double",
        "null"
      ],
      "default": 0
    }
    ...
}

Once again we can use the API to update the schema:

> curl -X PUT -H "X-Registry-Version: beta" \
               --data-binary @./schemas/beta/weather-schema.avsc \
               "http://localhost:8082/apis/registry/v2/groups/weather/artifacts/weather-avro-schema"

…and the Avro schema has now evolved and can be used by Consumers and Producers.

How About The Json Schema and Protobuf?

I was afraid you would ask me that…

The protobuf language guide suggests that adding a new optional visibilityDistance field is backward compatible (and it is accepted by Apicurio). Renaming a field should also be backward compatible and retiring a field by using the reserved notation but these are not accepted by Apicurio. So the following should, in my opinion, be backward compatible:

message Observations {
  reserved 8;  // Removed field
  optional double solarRadiation = 1; // Solar radiation
  optional double ultraViolet = 2; // Ultraviolet measurement
  optional double precipitationRate = 3; // Rate of rain
  optional double precipitationTotal24h = 4; // Total rain in preceding 24 hours
  optional double temperatureCelsius = 5; // Temperature in degrees Celsius
  optional double windChillCelsius = 6; // Experienced chill factor (in degress celsius)
  optional double windSpeed = 7; // Wind Speed (meters per second)
  optional double visibilityDistance = 9; // Visibility distance in metres
}

However updating the schema using the api:

> curl -X PUT -H "X-Registry-Version: beta" \
               --data-binary @./schemas/beta/weather-report.proto \
               "http://localhost:8082/apis/registry/v2/groups/weather/artifacts/weather-proto-schema"

Gives the message:

{"message": "Incompatible artifact: weather-proto-schema [PROTOBUF], num of incompatible diffs: {1}, list of diff types: [The new version of the protobuf artifact is not backward compatible. at /] Causes: The new version of the protobuf artifact is not backward compatible. at /"}

Which explains that a problem exists but not why (and after spending an evening with the source code I am not much wiser). A similar situation occurs with the updated Json schema.

These examples are not intended as a criticism of Apicurio (Occam’s razor indicates that it is simply my own misinterpretation of the schemas that is to blame), however are here to illustrate the complexity of schema versioning and how you will need to understand in depth how your chosen schema registry chooses to implement the compatibility level you have chosen.

A more subjective observation is that support for compatiblity for Avro is often more evolved in the Schema Registry.

Summary

Part three illustrates how we can use a Schema Registry to provide compatibility checks to reduce the risk of breaking the contract between Consumer and Producer. This increases our investment in our Schema Registry product:

• We need deep understanding of what compatibility guarantees our Schema Registry offers and how they are implemente
• We need to understand that versioning of a schema is a Schema Registry feature, not a schema framework feature
• We need to understand how our Schema Registry handles version ordering - in this case by upload time
• We may need to replicate or restore the schema version ordering, for example after a crash

The Schema Registry is not solely responsible for schema evolution. Preparing to evolve is essential - in our example the Consumer updated first - and this requires coordination often across departments.

There is a strong case for preparing for schema evolution as early as possible, preferably with the first version (for example evolving obligatory fields often produces challenging).

I hope this has given some insight into how the Schema Registry can help evolve your Schemas!