Kafka Connect and Schema Registry¶

📖 Kafka Connect - Connect framework documentation 📖 Schema Registry - Schema Registry documentation

Kafka Connect Overview¶

Architecture¶

📖 Connect Architecture - Connect architecture and concepts

Core Components: - Workers - JVM processes that execute connectors and tasks - Connectors - Logical jobs that define source/sink configuration - Tasks - Units of work that actually move data - Converters - Serialize/deserialize data between Connect and Kafka - Transforms - Optional record modifications (Single Message Transforms)

Worker Modes¶

Standalone Mode: - Single worker process - Configuration stored in local files - No fault tolerance or scalability - Good for development and testing - Started with: connect-standalone.sh worker.properties connector.properties

Distributed Mode: - Multiple worker processes form a Connect cluster - Configuration stored in Kafka topics (distributed, fault-tolerant) - Automatic task distribution and rebalancing - Managed via REST API - Started with: connect-distributed.sh worker.properties

📖 Distributed Mode - Distributed worker configuration

Worker Configuration¶

Key Settings: - bootstrap.servers - Kafka cluster address - group.id - Connect cluster group ID (distributed mode) - config.storage.topic - Topic for connector configurations - offset.storage.topic - Topic for source connector offsets - status.storage.topic - Topic for connector/task status - key.converter - Default key converter - value.converter - Default value converter - plugin.path - Directories for connector plugins

Converters¶

Important: Converters handle serialization between Connect's internal data format and Kafka. They are configured at the worker level (default) or per connector (override).

Connector Configuration¶

Source Connectors¶

• Read data from external systems into Kafka topics
• Track source offsets for exactly-once delivery
• Examples: JDBC Source, Debezium (CDC), FileStream Source

Sink Connectors¶

• Write data from Kafka topics to external systems
• Track consumer offsets for delivery guarantees
• Examples: JDBC Sink, Elasticsearch Sink, S3 Sink

Common Configuration Properties¶

{
  "name": "my-connector",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "tasks.max": "3",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://localhost:8081",
    "transforms": "route,timestamp",
    "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
    "transforms.route.regex": "(.*)",
    "transforms.route.replacement": "prefix-$1"
  }
}

Task Parallelism¶

• tasks.max - Maximum number of tasks for the connector
• Actual task count may be less than tasks.max (depends on data)
• Source connectors: tasks partition the data source
• Sink connectors: tasks are assigned Kafka partitions
• More tasks = higher parallelism (up to partition count for sinks)

Single Message Transforms (SMTs)¶

📖 SMT Documentation - Available transforms and configuration

Built-in Transforms¶

Transform Chain¶

{
  "transforms": "addTimestamp,route",
  "transforms.addTimestamp.type": "org.apache.kafka.connect.transforms.InsertField$Value",
  "transforms.addTimestamp.timestamp.field": "processed_at",
  "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
  "transforms.route.regex": "(.*)",
  "transforms.route.replacement": "processed-$1"
}

Error Handling¶

📖 Error Handling - Dead letter queues and error tolerance

Error Tolerance¶

• errors.tolerance=none (default) - Fail on first error
• errors.tolerance=all - Skip bad records, continue processing

Dead Letter Queue (DLQ)¶

{
  "errors.tolerance": "all",
  "errors.deadletterqueue.topic.name": "my-connector-dlq",
  "errors.deadletterqueue.topic.replication.factor": 3,
  "errors.deadletterqueue.context.headers.enable": true
}

Error Logging¶

{
  "errors.log.enable": true,
  "errors.log.include.messages": true
}

Connect REST API¶

📖 REST API Reference - Complete REST API documentation

Key Endpoints¶

Schema Registry¶

Overview¶

📖 Schema Registry Overview - Schema Registry fundamentals

• Centralized schema management for Kafka
• Stores schemas for Kafka topics (key and value)
• Enforces schema compatibility rules
• Supports Avro, Protobuf, and JSON Schema
• Schemas stored in the _schemas internal topic
• REST API for schema management

Subjects and Naming¶

Subject Naming Strategies: - TopicNameStrategy (default) - Subject name: <topic>-key / <topic>-value - RecordNameStrategy - Subject name: <fully-qualified-record-name> - TopicRecordNameStrategy - Subject name: <topic>-<fully-qualified-record-name>

📖 Subject Name Strategy - Configuring subject naming

Schema IDs¶

• Each schema version is assigned a globally unique integer ID
• IDs are sequential and never reused
• Producers include the schema ID in the message (first 5 bytes: magic byte + 4-byte ID)
• Consumers use the schema ID to retrieve the schema from the registry

Schema Compatibility¶

📖 Schema Evolution - Compatibility types and rules

Compatibility Types¶

Compatibility Rules for Avro¶

BACKWARD Compatible Changes: - Add a field with a default value - Remove a field (that had a default in old schema)

FORWARD Compatible Changes: - Remove a field that has a default value - Add a field (that has a default in new schema)

FULL Compatible Changes: - Add a field with a default value (when old schema also has default) - Remove a field that has a default value

Breaking Changes (all modes): - Change field type (int to string) - Rename a field without alias - Remove a required field (no default) - Add a required field (no default)

Compatibility Quick Reference¶

Schema Registry REST API¶

📖 Schema Registry API - Complete API reference

Key Endpoints¶

Schema Registration Example¶

# Register a new Avro schema
curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data '{"schema": "{\"type\":\"record\",\"name\":\"User\",\"fields\":[{\"name\":\"name\",\"type\":\"string\"},{\"name\":\"age\",\"type\":\"int\"}]}"}' \
  http://localhost:8081/subjects/users-value/versions

# Check compatibility
curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data '{"schema": "{\"type\":\"record\",\"name\":\"User\",\"fields\":[{\"name\":\"name\",\"type\":\"string\"},{\"name\":\"age\",\"type\":\"int\"},{\"name\":\"email\",\"type\":[\"null\",\"string\"],\"default\":null}]}"}' \
  http://localhost:8081/compatibility/subjects/users-value/versions/latest

Avro, Protobuf, and JSON Schema¶

Avro¶

• Binary format with schema
• Schema defined in JSON
• Compact serialization
• Rich type system (records, enums, arrays, maps, unions)
• Schema evolution with defaults and unions
• Most common format with Confluent Schema Registry

Protobuf¶

• Binary format defined by Google
• Schema defined in .proto files
• Language-neutral with code generation
• Backward compatible by design (field numbers)
• Good for polyglot environments

JSON Schema¶

• JSON format with schema validation
• Human-readable
• Larger payload size than Avro or Protobuf
• Good for debugging and web APIs
• Wide tooling support