Schemas in Connect

Summary

If you've taken the Kafka Fundamentals course, you'll recognize many concepts from lessons 4.3 and 4.4 on Schema and Evolution. We'll build on those foundations and show how Connect integrates with Schema Registry.

Let's start by understanding how Kafka Connect integrates with Schema Registry. Schema Registry is a separate service that stores and manages schemas for your data.

Here's how it works: When a source connector produces data, the converter checks if the schema is already registered. If not, it registers the schema and receives a unique schema ID. The converter then writes this schema ID along with the serialized data to Kafka.

On the consumer side, the sink connector's converter reads the schema ID from the message, fetches the corresponding schema from Schema Registry, and uses it to deserialize the data correctly.

This approach has a huge advantage: the schema is stored only once in Schema Registry, not in every message. This dramatically reduces message size while maintaining strong typing and schema evolution capabilities.

Note: Reference to Kafka Fundamentals

For a deeper dive into Schema Registry architecture and the fundamentals of schema evolution, see Kafka Fundamentals lessons 4.3 Schema Management and 4.4 Schema Evolution. Those lessons cover compatibility modes, schema versioning, and evolution strategies in detail.

Kafka Connect has its own internal schema representation that's converter-agnostic. This internal schema is what converters translate to and from.

A Connect schema defines the structure of your data. Each field has a name, a type like INT32 or STRING, whether it's required or optional, and optionally a default value.

Connect supports both primitive types like integers and strings, and complex types like structs, arrays, and maps. This rich type system allows you to model virtually any data structure.

Schema Evolution in Connect

Schema evolution is the ability to change schemas over time while maintaining compatibility with existing data. Connect supports several evolution strategies through Schema Registry.

Let me give you a concrete example. Suppose you start with this schema for a user record:

1{
2  "type": "record",
3  "name": "User",
4  "fields": [
5    {"name": "id", "type": "int"},
6    {"name": "name", "type": "string"}
7  ]
8}

Now you want to add an email field. Here's the evolved schema:

1{
2  "type": "record",
3  "name": "User",
4  "fields": [
5    {"name": "id", "type": "int"},
6    {"name": "name", "type": "string"},
7    {"name": "email", "type": ["null", "string"], "default": null}
8  ]
9}

Notice that email is optional with a default value of null. This makes the change backward compatible. Old consumers that don't know about the email field can still read the data, and new consumers can process records written with the old schema by using the default value.

Schema Registry enforces compatibility rules to prevent breaking changes. Here are the main compatibility modes.

BACKWARD compatibility means new schemas can read data written with old schemas. This is the default and most common mode. You can add optional fields or remove fields.

FORWARD compatibility means old schemas can read data written with new schemas. This is useful when you upgrade consumers before producers.

FULL compatibility means both backward and forward compatibility. This is the strictest and safest option for production systems.

NONE disables all compatibility checks. Use this only for development or when you have full control over all producers and consumers.

Configuring Schema Compatibility in Connect

You can set the compatibility mode when configuring your connector:

1# Connector configuration
2value.converter=io.confluent.connect.avro.AvroConverter
3value.converter.schema.registry.url=http://localhost:8081
4
5# For the subject (topic-value or topic-key)
6# Set via Schema Registry API or config

Compatibility is actually configured in Schema Registry per subject, not in the connector config itself. A subject is typically the topic name plus -key or -value.

You can set the compatibility mode using the Schema Registry REST API:

1# Set compatibility mode for a subject
2curl -X PUT http://localhost:8081/config/users-value \
3  -H "Content-Type: application/json" \
4  -d '{"compatibility": "FULL"}'

Schema Validation on Connector Start

When you start a connector with a schema-aware converter, Connect validates that the data structure matches the expected schema. If there's a mismatch, the connector will fail to start, alerting you to the problem before any bad data is produced.

This fail-fast behavior is much better than discovering data quality issues downstream in your pipeline.

Working with Schemaless Data

What if your source data doesn't have a schema? For example, raw log files or unstructured JSON? You have a few options:

First, you can use converters without schema support, like JsonConverter with schemas.enable set to false. This treats data as plain JSON without type information.

Second, you can define the schema in your connector configuration, forcing a structure onto unstructured data. This is common with SMTs that add or modify fields.

Third, you can write a custom converter that infers or applies schemas programmatically.

Best Practices for Schema Management

Always use schema-aware formats like Avro or Protobuf in production. The overhead is minimal compared to the benefits of type safety and evolution.

Set appropriate compatibility modes. FULL is safest, BACKWARD is most flexible for adding features over time.

Version your schemas explicitly and document changes. Even though Schema Registry tracks versions automatically, human documentation helps your team understand why changes were made.

Test schema changes in a development environment before deploying to production. Schema Registry allows you to test compatibility without actually registering schemas.

Monitor Schema Registry along with your Connect cluster. Schema registration failures can cause connector failures, so you want visibility into both systems.

Schema management is a critical aspect of building reliable data pipelines with Kafka Connect. By leveraging Schema Registry and following compatibility rules, you can evolve your data structures safely over time without breaking existing consumers.