Setup Kafka Schema Registry with Avro serialization for data processing

Install Java and dependencies

Schema Registry requires Java 11 or higher. Install the required dependencies first.

sudo apt update
sudo apt install -y openjdk-11-jdk wget curl unzip

sudo dnf update -y
sudo dnf install -y java-11-openjdk java-11-openjdk-devel wget curl unzip

Create dedicated user for Schema Registry

Run Schema Registry under a dedicated user account for security isolation.

sudo useradd --system --home-dir /opt/schema-registry --shell /bin/false --create-home schema-registry

Download and install Confluent Platform

Download the Confluent Platform which includes Schema Registry, or install just the Schema Registry component.

cd /tmp
wget https://packages.confluent.io/archive/7.5/confluent-7.5.0.tar.gz
tar -xzf confluent-7.5.0.tar.gz
sudo mv confluent-7.5.0 /opt/confluent
sudo chown -R schema-registry:schema-registry /opt/confluent

Configure Schema Registry properties

Create the main configuration file with connection details for your Kafka cluster.

# Schema Registry listeners
listeners=http://0.0.0.0:8081

Kafka cluster connection
kafkastore.bootstrap.servers=localhost:9092
kafkastore.topic=_schemas
kafkastore.topic.replication.factor=3

Schema Registry identity
schema.registry.group.id=schema-registry

Security settings
kafkastore.security.protocol=PLAINTEXT
schema.registry.inter.instance.protocol=http

Avro compatibility settings
avro.compatibility.level=BACKWARD
schema.compatibility.level=BACKWARD

Response cache settings
response.mediatype.preferred=application/vnd.schemaregistry.v1+json
response.mediatype.default=application/vnd.schemaregistry.v1+json

Configure Schema Registry with SSL

Enable SSL encryption for production deployments. Create SSL configuration.

# HTTPS listeners
listeners=https://0.0.0.0:8081

SSL keystore configuration
ssl.keystore.location=/opt/confluent/ssl/schema-registry.keystore.jks
ssl.keystore.password=changeme
ssl.key.password=changeme

SSL truststore configuration
ssl.truststore.location=/opt/confluent/ssl/schema-registry.truststore.jks
ssl.truststore.password=changeme

SSL client authentication
ssl.client.auth=required

Kafka SSL connection
kafkastore.bootstrap.servers=localhost:9093
kafkastore.security.protocol=SSL
kafkastore.ssl.keystore.location=/opt/confluent/ssl/client.keystore.jks
kafkastore.ssl.keystore.password=changeme
kafkastore.ssl.key.password=changeme
kafkastore.ssl.truststore.location=/opt/confluent/ssl/client.truststore.jks
kafkastore.ssl.truststore.password=changeme

Create SSL certificates

Generate SSL certificates for Schema Registry and client authentication.

sudo mkdir -p /opt/confluent/ssl
cd /opt/confluent/ssl

Generate CA certificate
sudo keytool -genkey -keyalg RSA -alias ca-cert -keystore ca.keystore.jks -validity 365 -dname "CN=example.com,OU=IT,O=Example,L=City,S=State,C=US" -storepass changeme -keypass changeme

Export CA certificate
sudo keytool -export -alias ca-cert -file ca-cert -keystore ca.keystore.jks -storepass changeme

Create truststore and import CA
sudo keytool -import -alias ca-cert -file ca-cert -keystore schema-registry.truststore.jks -storepass changeme -noprompt

Generate server certificate
sudo keytool -genkey -keyalg RSA -alias schema-registry -keystore schema-registry.keystore.jks -validity 365 -dname "CN=schema-registry.example.com,OU=IT,O=Example,L=City,S=State,C=US" -storepass changeme -keypass changeme

Sign server certificate with CA
sudo keytool -certreq -alias schema-registry -file schema-registry.csr -keystore schema-registry.keystore.jks -storepass changeme
sudo keytool -gencert -alias ca-cert -infile schema-registry.csr -outfile schema-registry.crt -keystore ca.keystore.jks -storepass changeme -validity 365
sudo keytool -import -alias ca-cert -file ca-cert -keystore schema-registry.keystore.jks -storepass changeme -noprompt
sudo keytool -import -alias schema-registry -file schema-registry.crt -keystore schema-registry.keystore.jks -storepass changeme

Set permissions
sudo chown -R schema-registry:schema-registry /opt/confluent/ssl
sudo chmod 600 /opt/confluent/ssl/*.jks

Create systemd service

Create a systemd service to manage Schema Registry startup and monitoring.

[Unit]
Description=Confluent Schema Registry
After=network.target
Requires=network.target

[Service]
Type=simple
User=schema-registry
Group=schema-registry
ExecStart=/opt/confluent/bin/schema-registry-start /opt/confluent/etc/schema-registry/schema-registry.properties
ExecStop=/opt/confluent/bin/schema-registry-stop
Restart=on-failure
RestartSec=10
StandardOutput=journal
StandardError=journal
SyslogIdentifier=schema-registry
KillMode=process
TimeoutStopSec=300

Security settings
NoNewPrivileges=yes
PrivateTmp=yes
ProtectSystem=strict
ProtectHome=yes
ReadWritePaths=/opt/confluent/logs

Environment
Environment=JAVA_HOME=/usr/lib/jvm/java-11-openjdk-amd64
Environment=SCHEMA_REGISTRY_HEAP_OPTS="-Xmx1G -Xms1G"
Environment=SCHEMA_REGISTRY_JVM_PERFORMANCE_OPTS="-server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35"

[Install]
WantedBy=multi-user.target

Configure log directory and permissions

Set up logging directory with correct ownership and permissions.

sudo mkdir -p /opt/confluent/logs
sudo chown schema-registry:schema-registry /opt/confluent/logs
sudo chmod 755 /opt/confluent/logs

Configure authentication

Enable basic authentication for Schema Registry access control.

# Enable authentication
authentication.method=BASIC
authentication.realm=SchemaRegistry
authentication.roles=admin,developer,readonly

JAAS configuration
schema.registry.resource.extension.class=io.confluent.kafka.schemaregistry.security.SchemaRegistrySecurityResourceExtension
confluent.schema.registry.auth.mechanism=JETTY_AUTH
schema.registry.auth.mechanism=JETTY_AUTH

Create user credentials file

Define users and their access levels for Schema Registry.

# Format: username: password,role1,role2
admin: admin123,admin
producer: producer123,developer
consumer: consumer123,readonly
app-user: app-secret,developer

Enable and start Schema Registry

Start the Schema Registry service and enable it to start on boot.

sudo systemctl daemon-reload
sudo systemctl enable schema-registry
sudo systemctl start schema-registry
sudo systemctl status schema-registry

Create sample Avro schema

Define an Avro schema for your data structure with proper field types and evolution support.

{
  "namespace": "com.example.users",
  "type": "record",
  "name": "User",
  "version": "1",
  "fields": [
    {
      "name": "id",
      "type": "long",
      "doc": "User unique identifier"
    },
    {
      "name": "username",
      "type": "string",
      "doc": "User login name"
    },
    {
      "name": "email",
      "type": "string",
      "doc": "User email address"
    },
    {
      "name": "created_at",
      "type": "long",
      "logicalType": "timestamp-millis",
      "doc": "User creation timestamp"
    },
    {
      "name": "metadata",
      "type": ["null", {
        "type": "map",
        "values": "string"
      }],
      "default": null,
      "doc": "Optional user metadata"
    }
  ]
}

Register schema with Schema Registry

Upload the schema to Schema Registry using the REST API.

# Register schema for a subject (topic)
curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data @/tmp/user-schema.json \
  http://localhost:8081/subjects/users-value/versions

Register with authentication
curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -u admin:admin123 \
  --data '{"schema": "{\"namespace\": \"com.example.users\", \"type\": \"record\", \"name\": \"User\", \"fields\": [{\"name\": \"id\", \"type\": \"long\"}, {\"name\": \"username\", \"type\": \"string\"}, {\"name\": \"email\", \"type\": \"string\"}, {\"name\": \"created_at\", \"type\": \"long\", \"logicalType\": \"timestamp-millis\"}, {\"name\": \"metadata\", \"type\": [\"null\", {\"type\": \"map\", \"values\": \"string\"}], \"default\": null}]}"}' \
  http://localhost:8081/subjects/users-value/versions

Java producer configuration

Configure a Kafka producer to use Avro serialization with Schema Registry.

# Kafka cluster connection
bootstrap.servers=localhost:9092

Schema Registry settings
schema.registry.url=http://localhost:8081
schema.registry.basic.auth.credentials.source=USER_INFO
schema.registry.basic.auth.user.info=producer:producer123

Avro serialization
key.serializer=org.apache.kafka.common.serialization.StringSerializer
value.serializer=io.confluent.kafka.serializers.KafkaAvroSerializer

Producer settings
acks=all
retries=2147483647
max.in.flight.requests.per.connection=1
enable.idempotence=true
compression.type=snappy

Schema evolution
auto.register.schemas=false
use.latest.version=true

Java consumer configuration

Configure a Kafka consumer to deserialize Avro messages using Schema Registry.

# Kafka cluster connection
bootstrap.servers=localhost:9092

Schema Registry settings
schema.registry.url=http://localhost:8081
schema.registry.basic.auth.credentials.source=USER_INFO
schema.registry.basic.auth.user.info=consumer:consumer123

Avro deserialization
key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
value.deserializer=io.confluent.kafka.serializers.KafkaAvroDeserializer

Consumer settings
group.id=user-consumer-group
auto.offset.reset=earliest
enable.auto.commit=false

Avro specific settings
specific.avro.reader=true

Python client configuration

Configure Python producer and consumer using confluent-kafka with Avro support.

# Install Python dependencies
pip install confluent-kafka[avro] requests

from confluent_kafka import avro
from confluent_kafka.avro import AvroProducer
from confluent_kafka.schema_registry import SchemaRegistryClient
import json

Schema Registry configuration
schema_registry_conf = {
    'url': 'http://localhost:8081',
    'basic.auth.user.info': 'producer:producer123'
}

Producer configuration
producer_conf = {
    'bootstrap.servers': 'localhost:9092',
    'schema.registry.url': 'http://localhost:8081',
    'schema.registry.basic.auth.credentials.source': 'USER_INFO',
    'schema.registry.basic.auth.user.info': 'producer:producer123'
}

Load schema
with open('/tmp/user-schema.json', 'r') as f:
    user_schema = json.load(f)

Create producer
producer = AvroProducer(producer_conf, default_value_schema=avro.loads(json.dumps(user_schema)))

Send message
user_data = {
    'id': 12345,
    'username': 'john_doe',
    'email': 'john@example.com',
    'created_at': 1703097600000,
    'metadata': {'role': 'user', 'department': 'engineering'}
}

producer.produce(topic='users', value=user_data, key='user_12345')
producer.flush()

Enable compatibility checking

Configure compatibility levels to control schema evolution policies.

# Set global compatibility level
curl -X PUT -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -u admin:admin123 \
  --data '{"compatibility": "BACKWARD"}' \
  http://localhost:8081/config

Set subject-specific compatibility
curl -X PUT -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -u admin:admin123 \
  --data '{"compatibility": "BACKWARD_TRANSITIVE"}' \
  http://localhost:8081/config/users-value

Create evolved schema version

Add a new optional field to demonstrate backward-compatible schema evolution.

{
  "namespace": "com.example.users",
  "type": "record",
  "name": "User",
  "version": "2",
  "fields": [
    {
      "name": "id",
      "type": "long",
      "doc": "User unique identifier"
    },
    {
      "name": "username",
      "type": "string",
      "doc": "User login name"
    },
    {
      "name": "email",
      "type": "string",
      "doc": "User email address"
    },
    {
      "name": "created_at",
      "type": "long",
      "logicalType": "timestamp-millis",
      "doc": "User creation timestamp"
    },
    {
      "name": "last_login",
      "type": ["null", "long"],
      "logicalType": "timestamp-millis",
      "default": null,
      "doc": "Last login timestamp (new field)"
    },
    {
      "name": "metadata",
      "type": ["null", {
        "type": "map",
        "values": "string"
      }],
      "default": null,
      "doc": "Optional user metadata"
    }
  ]
}

Register evolved schema

Upload the new schema version and verify compatibility.

# Test compatibility before registering
curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -u admin:admin123 \
  --data @/tmp/user-schema-v2.json \
  http://localhost:8081/compatibility/subjects/users-value/versions/latest

Register new version if compatible
curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -u admin:admin123 \
  --data @/tmp/user-schema-v2.json \
  http://localhost:8081/subjects/users-value/versions