Using Schema Registry
1. Understanding Schema Registry Role
| Aspect | Description | Impact |
|---|---|---|
| Schema Evolution | Enables controlled changes to schema definitions over time | Ensures backward/forward compatibility in data processing |
| Data Validation | Validates incoming data against registered schemas | Maintains data integrity and prevents malformed messages |
| Versioning | Manages multiple versions of schemas with strict rules | Supports migration paths for consumers and producers |
| Serialization/Deserialization | Handles encoding and decoding of messages using schemas | Standardizes data format across services |
Example: Schema Registration
{
"subject": "user-avro",
"version": 1,
"schema": "{\n \"type\": \"record\",\n \"name\": \"User\",\n \"fields\": [\n {\"name\": \"id\", \"type\": \"int\"},\n {\"name\": \"name\", \"type\": \"string\"}\n ]\n}"
}Note: Schema Registry ensures that producers and consumers are always aligned on the structure of messages, preventing runtime errors due to schema mismatches.
2. Installing Schema Registry
Example: Installing Schema Registry on Ubuntu
# Add Confluent repository
wget -qO - https://packages.confluent.io/deb/7.4/archive.pub | sudo apt-key add -
echo "deb [arch=amd64] https://packages.confluent.io/deb/7.4/archive stable main" | sudo tee /etc/apt/sources.list.d/confluent.list
# Update package list and install Schema Registry
sudo apt-get update
sudo apt-get install confluent-schema-registry| Component | Description | Installation Method |
|---|---|---|
| Schema Registry | Manages and stores schemas for Kafka topics | Package manager (apt, yum), Docker, or manual download |
| Kafka Cluster | Required for Schema Registry to connect to Kafka | Must be running before installing Schema Registry |
Note: Schema Registry requires a running Kafka cluster and ZooKeeper instance to function properly.
Warning: Ensure that the version of Schema Registry matches your Kafka version to avoid compatibility issues.
Installation Steps
- Install Confluent Platform (includes Schema Registry)
- Start Kafka and ZooKeeper services
- Configure Schema Registry properties (e.g.,
schema.registry.host.name) - Start Schema Registry service
3. Configuring Schema Registry URL
| Property | Default Value | Description |
|---|---|---|
| schema.registry.url | http://localhost:8081 |
Base URL of the Schema Registry server |
| schema.registry.urls | http://localhost:8081 |
Comma-separated list of Schema Registry URLs (for failover) |
Example: Producer Configuration
{
"bootstrap.servers": "localhost:9092",
"key.serializer": "org.apache.kafka.common.serialization.StringSerializer",
"value.serializer": "io.confluent.kafka.serializers.KafkaAvroSerializer",
"schema.registry.url": "http://localhost:8081"
}Example: Consumer Configuration
{
"bootstrap.servers": "localhost:9092",
"key.deserializer": "org.apache.kafka.common.serialization.StringDeserializer",
"value.deserializer": "io.confluent.kafka.serializers.KafkaAvroDeserializer",
"schema.registry.url": "http://localhost:8081"
}4. Registering Schemas
| Field | Description | Example |
|---|---|---|
| subject | The subject name for the schema | user-avro |
| schema | The schema definition in JSON format | {"type": "record", "name": "User", "fields": [{"name": "id", "type": "int"}, {"name": "name", "type": "string"}]} |
| schemaType | Schema type (AVRO, JSON, PROTOBUF) | AVRO |
| version | Version of the schema (auto-incremented) | 1 |
Example: Registering a new schema
curl -X POST http://localhost:8081/subjects/user-avro/versions \
-H "Content-Type: application/vnd.schemaregistry.v1+json" \
-d '{"schema": "{\"type\": \"record\", \"name\": \"User\", \"fields\": [{\"name\": \"id\", \"type\": \"int\"}, {\"name\": \"name\", \"type\": \"string\"}]}"}'
Note: The Schema Registry will return the registered schema ID and version upon successful registration.
5. Retrieving Schemas
| Method | Description | Parameters | Response |
|---|---|---|---|
| GET /schemas/{id} | Retrieve a schema by its ID | id: Schema ID (string) |
Schema content in JSON format |
| GET /schemas/{id}/versions | Get all versions of a schema by ID | id: Schema ID (string) |
List of versions (array of integers) |
| GET /schemas/{id}/versions/{version} | Retrieve a specific version of a schema | id: Schema ID (string)version: Version number (integer) |
Schema content in JSON format |
| GET /subjects | List all registered subjects | None | List of subject names (array of strings) |
| GET /subjects/{subject}/versions | Get all versions of a subject | subject: Subject name (string) |
List of versions (array of integers) |
| GET /subjects/{subject}/versions/{version} | Retrieve a specific version of a subject | subject: Subject name (string)version: Version number (integer) |
Schema content in JSON format |
Note: Schema IDs are auto-generated and unique within the registry.
Warning: Schema versions are not guaranteed to be sequential.
6. Listing Subjects
| Property | Description | Example |
|---|---|---|
| subject | Name of the schema subject | user-avro |
| version | Schema version number | 1 |
| id | Unique identifier for the schema | 12345 |
Example: List all versions of a subject
curl -X GET http://localhost:8081/subjects/user-avro/versions
7. Deleting Schema Versions
| Version | Action | Notes |
|---|---|---|
| 1.0 | DELETE /subjects/{subject}/versions/{version} |
Deletes a specific schema version from the registry |
| 2.0 | DELETE /subjects/{subject}/versions/{version}?permanent=true |
Permanently deletes a schema version, making it unrecoverable |
Note: Deleting a schema version does not affect the subject's compatibility settings or other versions.
Warning: Permanently deleted schema versions cannot be recovered.
8. Setting Compatibility Mode
| Mode | Description | Notes |
|---|---|---|
| BACKWARD | Allows new schemas to be compatible with older schemas | Ensures that producers can send data using new schema and consumers can read it with older schema |
| FORWARD | Allows older schemas to be compatible with newer schemas | Ensures that consumers can read data using older schema and producers can send it with newer schema |
| FULL | Ensures full compatibility in both directions | Requires that schemas are fully compatible in both backward and forward direction |
| BACKWARD_TRANSITIVE | Allows new schemas to be compatible with older schemas and transitive compatibility | Enforces backward compatibility including all previous versions in the schema history |
| FORWARD_TRANSITIVE | Allows older schemas to be compatible with newer schemas and transitive compatibility | Enforces forward compatibility including all previous versions in the schema history |
| FULL_TRANSITIVE | Ensures full transitive compatibility in both directions | Enforces full compatibility including all previous versions in the schema history |
Example: Configuring Compatibility Mode via REST API
curl -X PUT http://localhost:8081/subjects/my-topic-value/versions/latest
-H "Content-Type: application/vnd.schemaregistry.v1+json"
--data "{\"compatibility\": \"BACKWARD\"}"
Note: The compatibility mode is set per subject (topic) and can be overridden at the schema level.
9. Checking Compatibility
| Compatibility Type | Description | API Endpoint | Notes |
|---|---|---|---|
| Backward Compatibility | Checks if a new schema can read data produced by the previous version | /compatibility/subjects/{subject}/versions/{version} |
Default mode for most use cases |
| Forward Compatibility | Ensures that data produced by the new schema can be read by previous versions | /compatibility/subjects/{subject}/versions/{version} |
Use when evolving schemas in a way that supports older consumers |
| Full Compatibility | Ensures both backward and forward compatibility | /compatibility/subjects/{subject}/versions/{version} |
Most restrictive, ensures full schema evolution safety |
| None Compatibility | No compatibility checks performed | /compatibility/subjects/{subject}/versions/{version} |
Use with caution, not recommended for production |
Example: Checking Schema Compatibility
curl -X GET \
http://localhost:8081/compatibility/subjects/user-schema/versions/1 \
-H 'Content-Type: application/vnd.schemaregistry.v1+json'
Note: The Schema Registry supports multiple compatibility levels that can be configured per subject or globally.
Warning: If compatibility checks fail, the new schema will not be registered and the operation will be rejected.
10. Exporting Schemas
Example: Export Schema to File
curl -X GET \
http://schema-registry-host:8081/subjects/my-topic-value/versions/1 \
-H 'Accept: application/vnd.schemaregistry.v1+json' \
> exported-schema.json| Option | Values | Description |
|---|---|---|
| Subject Name | my-topic-value |
Name of the schema subject to export |
| Version | 1 |
Schema version to export |
| Accept Header | application/vnd.schemaregistry.v1+json |
Specifies the response format |
Note: The exported schema will include the full schema definition in JSON format.