10.2.2 – Avro Serialization with Kafka

Module: Section 10 – Kafka Ecosystem & Tools
Section: 10.2 – Avro & Strongly Typed Messaging
Goal: Learn how to use KafkaAvroSerializer and KafkaAvroDeserializer with Schema Registry

Introduction

Kafka's KafkaAvroSerializer and KafkaAvroDeserializer integrate seamlessly with Schema Registry to provide automatic schema registration, retrieval, and validation.

Why This Module Matters: These serializers handle all Schema Registry interactions automatically, making Avro integration straightforward.

Kafka Avro Serialization Architecture

Producer Application
    ↓ (Avro record)
KafkaAvroSerializer
    ↓ (checks Schema Registry)
Schema Registry (returns schema_id)
    ↓ (prepends schema_id to message)
Kafka Broker
    ↓ (stores: [schema_id][binary_data])
    ↓
Kafka Consumer
    ↓ (receives: [schema_id][binary_data])
KafkaAvroDeserializer
    ↓ (retrieves schema from Schema Registry)
Schema Registry (returns schema for schema_id)
    ↓ (deserializes using schema)
Consumer Application (receives Avro record)

Producer Setup

Maven Dependencies

<dependencies>
  <!-- Kafka Clients -->
  <dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.6.1</version>
  </dependency>

  <!-- Avro -->
  <dependency>
    <groupId>org.apache.avro</groupId>
    <artifactId>avro</artifactId>
    <version>1.11.3</version>
  </dependency>

  <!-- Kafka Avro Serializer (includes Schema Registry client) -->
  <dependency>
    <groupId>io.confluent</groupId>
    <artifactId>kafka-avro-serializer</artifactId>
    <version>7.5.0</version>
  </dependency>
</dependencies>

<repositories>
  <!-- Confluent repository for Schema Registry libraries -->
  <repository>
    <id>confluent</id>
    <url>https://packages.confluent.io/maven/</url>
  </repository>
</repositories>

Producer Configuration

import org.apache.kafka.clients.producer.*;
import io.confluent.kafka.serializers.KafkaAvroSerializer;
import java.util.Properties;

Properties props = new Properties();

// Kafka broker configuration
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.CLIENT_ID_CONFIG, "avro-producer");

// Avro serialization
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, 
    "org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, 
    KafkaAvroSerializer.class.getName());

// Schema Registry configuration
props.put("schema.registry.url", "http://localhost:8081");

// Optional: Auto-register schema (default: true)
props.put("auto.register.schemas", "true");

// Optional: Use latest version for validation (default: true)
props.put("use.latest.version", "true");

KafkaProducer<String, GenericRecord> producer = new KafkaProducer<>(props);

Key Configuration Properties

Producing Avro Messages

Using GenericRecord (Dynamic Schema)

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;

// 1. Define Avro schema
String schemaString = """
{
  "type": "record",
  "name": "User",
  "namespace": "com.example.avro",
  "fields": [
    {"name": "userId", "type": "long"},
    {"name": "username", "type": "string"},
    {"name": "email", "type": ["null", "string"], "default": null}
  ]
}
""";

Schema schema = new Schema.Parser().parse(schemaString);

// 2. Create Avro record
GenericRecord user = new GenericData.Record(schema);
user.put("userId", 123L);
user.put("username", "alice");
user.put("email", "alice@example.com");

// 3. Send to Kafka
ProducerRecord<String, GenericRecord> record = 
    new ProducerRecord<>("users", "user-123", user);

producer.send(record, (metadata, exception) -> {
    if (exception == null) {
        System.out.printf("Sent to partition %d, offset %d\n", 
            metadata.partition(), metadata.offset());
    } else {
        exception.printStackTrace();
    }
});

producer.flush();

What Happens Internally

1. KafkaAvroSerializer receives GenericRecord
   ↓
2. Extract schema from record
   ↓
3. Compute schema hash
   ↓
4. Check local cache: Is schema_id known?
   ↓
   NO → Continue to step 5
   ↓
5. POST /subjects/users-value/versions
   {
     "schema": "{\"type\":\"record\", ...}"
   }
   ↓
6. Schema Registry:
   - Validates syntax
   - Checks compatibility with previous versions
   - If compatible: Assigns schema_id (e.g., 42)
   - Writes to _schemas topic
   - Returns {"id": 42}
   ↓
7. Serializer caches schema_id = 42
   ↓
8. Prepend magic byte (0x00) + schema_id (4 bytes)
   ↓
9. Serialize Avro record to binary
   ↓
10. Final message: [0x00][0x00 0x00 0x00 0x2A][binary_data]
   ↓
11. Send to Kafka broker

Consumer Setup

Consumer Configuration

import org.apache.kafka.clients.consumer.*;
import io.confluent.kafka.serializers.KafkaAvroDeserializer;
import java.util.Properties;
import java.util.Collections;
import java.time.Duration;

Properties props = new Properties();

// Kafka broker configuration
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "avro-consumer-group");
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

// Avro deserialization
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, 
    "org.apache.kafka.common.serialization.StringDeserializer");
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, 
    KafkaAvroDeserializer.class.getName());

// Schema Registry configuration
props.put("schema.registry.url", "http://localhost:8081");

// CRITICAL: Use reader schema (not writer schema)
props.put("specific.avro.reader", "false");  // GenericRecord
// OR
props.put("specific.avro.reader", "true");   // SpecificRecord (code-generated)

KafkaConsumer<String, GenericRecord> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("users"));

Consuming Avro Messages

while (true) {
    ConsumerRecords<String, GenericRecord> records = 
        consumer.poll(Duration.ofMillis(100));

    for (ConsumerRecord<String, GenericRecord> record : records) {
        GenericRecord user = record.value();
        
        // Extract fields
        long userId = (Long) user.get("userId");
        String username = (String) user.get("username");
        String email = (String) user.get("email");
        
        System.out.printf("Received user: id=%d, username=%s, email=%s\n", 
            userId, username, email);
    }
}

What Happens Internally

1. Consumer receives message: [0x00][0x00 0x00 0x00 0x2A][binary_data]
   ↓
2. KafkaAvroDeserializer extracts schema_id (42) from bytes 1-4
   ↓
3. Check local cache: Is schema known for id=42?
   ↓
   NO → Continue to step 4
   ↓
4. GET /schemas/ids/42
   ↓
5. Schema Registry returns schema JSON
   ↓
6. Deserializer caches schema for id=42
   ↓
7. Decode binary data using schema
   ↓
8. Return GenericRecord to consumer

Schema Registry Integration Details

Automatic Schema Registration

Producer:

// First message with new schema
producer.send(new ProducerRecord<>("users", user1));
// → Schema Registry: POST /subjects/users-value/versions → Returns id=1

// Second message with same schema
producer.send(new ProducerRecord<>("users", user2));
// → No Schema Registry call (cached)

// Third message with evolved schema (new field added)
producer.send(new ProducerRecord<>("users", user3));
// → Schema Registry: POST /subjects/users-value/versions → Returns id=2

Subject Naming Strategy

Default Strategy: TopicNameStrategy

Topic: users
Key Subject:   users-key
Value Subject: users-value

Alternative: RecordNameStrategy

Topic: events
Value Subject: com.example.avro.User (based on record name)

Alternative: TopicRecordNameStrategy

Topic: events
Value Subject: events-com.example.avro.User

Configuration:

props.put("value.subject.name.strategy", 
    "io.confluent.kafka.serializers.subject.TopicNameStrategy"); // Default

Handling Schema Evolution

Adding Optional Field (Backward Compatible)

Producer v1:

Schema v1 = new Schema.Parser().parse("""
{
  "type": "record",
  "name": "User",
  "fields": [
    {"name": "userId", "type": "long"},
    {"name": "username", "type": "string"}
  ]
}
""");

Producer v2 (add optional email):

Schema v2 = new Schema.Parser().parse("""
{
  "type": "record",
  "name": "User",
  "fields": [
    {"name": "userId", "type": "long"},
    {"name": "username", "type": "string"},
    {"name": "email", "type": ["null", "string"], "default": null}
  ]
}
""");

Consumer Behavior:

• ✅ Old consumer (v1) reading new data (v2): Ignores email field
• ✅ New consumer (v2) reading old data (v1): Uses default (null) for email

Error Handling

SerializationException: Schema Incompatible

try {
    producer.send(new ProducerRecord<>("users", user));
} catch (SerializationException e) {
    // Schema incompatible with previous versions
    System.err.println("Schema validation failed: " + e.getMessage());
    // e.g., "Schema being registered is incompatible with an earlier schema"
}

Causes:

• Removed required field
• Changed field type (non-promotable)
• Renamed field
• Violated compatibility mode

RestClientException: Schema Registry Unreachable

try {
    producer.send(new ProducerRecord<>("users", user));
} catch (RestClientException e) {
    // Schema Registry unavailable
    System.err.println("Schema Registry error: " + e.getMessage());
    // e.g., "Connection refused: http://localhost:8081"
}

Solutions:

• Check Schema Registry URL
• Verify Schema Registry is running
• Check network connectivity
• Review authentication credentials

Performance Optimization

Schema Caching

Producer/Consumer Cache:

// Default cache size: 1000 schemas
props.put("max.schemas.per.subject", "1000");

Why It Matters:

• First message: 20ms latency (Schema Registry call)
• Cached messages: 0.1ms latency (no network call)
• 99.99% cache hit rate in production

Connection Pooling

Schema Registry Client Configuration:

// Connection pool size (default: 25)
props.put("schema.registry.max.connections.per.host", "50");

// Connection timeout (default: 30000ms)
props.put("schema.registry.request.timeout.ms", "10000");

Security Configuration

Basic Authentication

props.put("schema.registry.url", "https://schema-registry.company.com");
props.put("basic.auth.credentials.source", "USER_INFO");
props.put("basic.auth.user.info", "username:password");

SSL/TLS

props.put("schema.registry.url", "https://schema-registry.company.com:8081");
props.put("schema.registry.ssl.truststore.location", "/path/to/truststore.jks");
props.put("schema.registry.ssl.truststore.password", "truststore-password");

Key Takeaways

1. KafkaAvroSerializer: Automatically registers schemas and prepends schema ID to messages.

2. KafkaAvroDeserializer: Retrieves schemas from Schema Registry and deserializes binary data.

3. Configuration: Requires schema.registry.url, uses auto.register.schemas (default: true).

4. Message Format: [0x00][schema_id (4 bytes)][avro_binary_data]

5. Schema Registry Integration: Serializer/deserializer handle all REST API calls automatically.

6. Caching: 99.99% cache hit rate makes Schema Registry overhead negligible.

7. Subject Naming: Default is <topic>-key and <topic>-value.

8. Error Handling: Catch SerializationException (compatibility) and RestClientException (connectivity).

9. Performance: First message ~20ms, cached messages ~0.1ms.

10. Security: Supports basic auth and SSL/TLS for Schema Registry.

What's Next?

Next Module: 10.2.3 – SpecificRecord vs GenericRecord

What We Covered:

• ✅ KafkaAvroSerializer/Deserializer setup
• ✅ Producer/consumer configuration
• ✅ Schema Registry integration
• ✅ Error handling
• ✅ Performance optimization

What's Coming:

• ❌ Code generation with SpecificRecord
• ❌ GenericRecord vs SpecificRecord tradeoffs
• ❌ Type safety patterns

References:

• Schema Registry → 10.1: Schema Registry Foundations
• Schema evolution → 10.3: Schema Evolution & Compatibility
• Operations → 10.4: Schema Registry Operations