3.1 Serialization & Data Modeling

Module Overview

Goal: Separate data correctness from producer mechanics. Learn how to properly serialize data for Kafka.

What You'll Learn:

• Why serialization matters in distributed systems
• Key vs value serialization
• Built-in serializers
• Custom serializers
• Schema evolution strategies (Avro, JSON Schema, Protobuf)
• Common serialization failures and how to avoid them
• Best practices for data modeling

What's NOT Covered Here:

• Producer configuration details (covered in Module 2.1)
• Partitioning logic (covered in Module 4.1)
• Schema Registry operations (separate topic)

Key Insight: Serialization problems are data problems, not producer problems — mixing them causes confusion.

1. Why Serialization Matters

The Fundamental Problem

Kafka stores and transmits bytes. Your application works with objects.

Application Object → [Serializer] → Bytes → Kafka
Kafka → Bytes → [Deserializer] → Application Object

Serialization: Object → Bytes (Producer side)
Deserialization: Bytes → Object (Consumer side)

What Can Go Wrong?

// Producer sends
producer.send(new ProducerRecord<>("topic", user));

// Consumer expects User but gets...?
User user = consumer.poll(); // What if format changed?

Common Issues:

• Schema changes break consumers
• Type mismatches (producer sends String, consumer expects Integer)
• Version incompatibility
• Data corruption
• Performance degradation (inefficient formats)

2. Serializer Fundamentals

The Serializer Interface

public interface Serializer<T> {
    byte[] serialize(String topic, T data);
    
    // Optional lifecycle methods
    void configure(Map<String, ?> configs, boolean isKey);
    void close();
}

Key Points:

• Converts type T to byte[]
• Receives topic name (for topic-specific logic)
• Separate serializers for keys and values

Configuration

Properties props = new Properties();

// Key serializer
props.put("key.serializer", 
    "org.apache.kafka.common.serialization.StringSerializer");

// Value serializer
props.put("value.serializer", 
    "org.apache.kafka.common.serialization.StringSerializer");

Why Separate?

• Keys and values can be different types
• Key determines partition (needs consistent serialization)
• Value is payload (can be more flexible)

3. Built-in Serializers

Kafka provides serializers for common Java types.

StringSerializer

props.put("value.serializer", 
    "org.apache.kafka.common.serialization.StringSerializer");

// Usage
ProducerRecord<String, String> record = 
    new ProducerRecord<>("topic", "key", "value");
producer.send(record);

Encoding: UTF-8 by default

Use For:

• Text data
• JSON strings (serialize object to JSON string first)
• Simple key-value pairs

IntegerSerializer / LongSerializer

props.put("key.serializer", 
    "org.apache.kafka.common.serialization.IntegerSerializer");

// Usage
ProducerRecord<Integer, String> record = 
    new ProducerRecord<>("topic", 12345, "value");

Use For:

• Numeric keys (user IDs, order IDs)
• Counters
• Timestamps (as Long)

ByteArraySerializer

props.put("value.serializer", 
    "org.apache.kafka.common.serialization.ByteArraySerializer");

// Usage
byte[] data = someObject.toBytes();
ProducerRecord<String, byte[]> record = 
    new ProducerRecord<>("topic", "key", data);

Use For:

• Already serialized data
• Binary formats (images, files)
• Custom binary protocols
• Protobuf/Avro (serialize first, then send bytes)

ByteBufferSerializer / BytesSerializer

props.put("value.serializer", 
    "org.apache.kafka.common.serialization.ByteBufferSerializer");

Use For:

• Zero-copy scenarios
• Working with NIO ByteBuffers

Complete List of Built-in Serializers

Kafka provides these serializers out of the box:

Package: org.apache.kafka.common.serialization

3.5 Pluggable Serializers: The Serializer Architecture

Why Serializers are Pluggable

Kafka's serializer design is pluggable — you can swap serializers without changing producer code.

// Same producer code, different serializers
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
// OR
props.put("value.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer");
// OR
props.put("value.serializer", "com.mycompany.MyCustomSerializer");

Benefits:

1. Separation of concerns: Producer logic separate from serialization
2. Easy migration: Change format without rewriting producer
3. Reusability: Same serializer across multiple producers
4. Testability: Mock serializers for testing

How Pluggable Serializers Work

1. Producer Loads Serializer at Startup

Properties props = new Properties();
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

// Producer instantiates the serializer via reflection
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
// → Loads StringSerializer class
// → Calls configure()

2. Producer Calls Serializer on Every Send

producer.send(new ProducerRecord<>("topic", "key", "value"));

// Internally:
// 1. keySerializer.serialize("topic", "key")   → byte[]
// 2. valueSerializer.serialize("topic", "value") → byte[]
// 3. Send bytes to Kafka

3. Serializer Lifecycle

CREATE → CONFIGURE → SERIALIZE (many times) → CLOSE
   │         │             │                     │
   │         │             │                     └─ Release resources
   │         │             └─ Convert objects to bytes
   │         └─ Initialize with configs
   └─ Instantiate via reflection

Using Third-Party Serializers

Example: Confluent Avro Serializer

<!-- Maven dependency -->
<dependency>
  <groupId>io.confluent</groupId>
  <artifactId>kafka-avro-serializer</artifactId>
  <version>7.5.0</version>
</dependency>

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer");

// Avro-specific config
props.put("schema.registry.url", "http://localhost:8081");

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

// Send Avro record
GenericRecord avroRecord = ...;
producer.send(new ProducerRecord<>("topic", "key", avroRecord));

The producer doesn't know or care that it's Avro — it just calls serialize().

Example: Protobuf Serializer

<dependency>
  <groupId>io.confluent</groupId>
  <artifactId>kafka-protobuf-serializer</artifactId>
  <version>7.5.0</version>
</dependency>

props.put("value.serializer", "io.confluent.kafka.serializers.protobuf.KafkaProtobufSerializer");
props.put("schema.registry.url", "http://localhost:8081");

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

Example: JSON Serializer

<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-clients</artifactId>
  <version>3.7.0</version>
</dependency>

// Kafka doesn't include a built-in JSON serializer, but you can use:
// 1. Spring Kafka's JsonSerializer
// 2. Write your own (see Custom Serializers section)
// 3. Use StringSerializer + manual JSON conversion

// Option 1: Spring Kafka
props.put("value.serializer", "org.springframework.kafka.support.serializer.JsonSerializer");

// Option 2: Manual JSON with StringSerializer
ObjectMapper mapper = new ObjectMapper();
String json = mapper.writeValueAsString(myObject);
producer.send(new ProducerRecord<>("topic", "key", json));

Advantages of Pluggable Design

1. Easy Format Migration

// Phase 1: JSON
props.put("value.serializer", "com.mycompany.JsonSerializer");

// Phase 2: Migrate to Avro (no producer code changes!)
props.put("value.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer");
props.put("schema.registry.url", "http://localhost:8081");

2. Environment-Specific Serializers

// Development: Human-readable JSON
if (env.equals("dev")) {
    props.put("value.serializer", "com.mycompany.JsonSerializer");
}

// Production: Efficient Avro
if (env.equals("prod")) {
    props.put("value.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer");
}

3. Testing with Mock Serializers

// Test with a no-op serializer
public class NoOpSerializer implements Serializer<String> {
    public byte[] serialize(String topic, String data) {
        return data.getBytes(); // Simple pass-through
    }
}

props.put("value.serializer", "com.mycompany.test.NoOpSerializer");

Best Practices for Pluggable Serializers

✅ DO:

• Keep serializers stateless (thread-safe)
• Make serializers configurable via configure()
• Handle null values gracefully
• Release resources in close()
• Document expected configuration keys

❌ DON'T:

• Put business logic in serializers
• Make serializers depend on external state
• Throw unchecked exceptions (wrap in SerializationException)
• Ignore the topic parameter (useful for topic-specific logic)

4. Custom Serializers

When to Create Custom Serializers

✅ Good Reasons:

• Domain objects that don't fit standard serializers
• Custom binary formats
• Tight integration with existing serialization libraries
• Performance optimization for specific types

❌ Bad Reasons:

• Wrapping JSON serialization (just use JsonSerializer)
• Adding business logic (serializers should be pure)

Creating a Custom Serializer

Example: UserSerializer

public class User {
    private long id;
    private String name;
    private String email;
    
    // Constructor, getters, setters
}

import org.apache.kafka.common.serialization.Serializer;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;

public class UserSerializer implements Serializer<User> {
    
    @Override
    public byte[] serialize(String topic, User user) {
        if (user == null) {
            return null;
        }
        
        byte[] nameBytes = user.getName().getBytes(StandardCharsets.UTF_8);
        byte[] emailBytes = user.getEmail().getBytes(StandardCharsets.UTF_8);
        
        // Calculate total size: 8 (id) + 4 (name length) + name + 4 (email length) + email
        ByteBuffer buffer = ByteBuffer.allocate(
            8 + 4 + nameBytes.length + 4 + emailBytes.length
        );
        
        buffer.putLong(user.getId());
        
        buffer.putInt(nameBytes.length);
        buffer.put(nameBytes);
        
        buffer.putInt(emailBytes.length);
        buffer.put(emailBytes);
        
        return buffer.array();
    }
    
    @Override
    public void close() {
        // Cleanup if needed
    }
}

Usage:

props.put("value.serializer", "com.myapp.UserSerializer");

User user = new User(123L, "Alice", "alice@example.com");
ProducerRecord<String, User> record = 
    new ProducerRecord<>("users", String.valueOf(user.getId()), user);
producer.send(record);

Custom Serializer Best Practices

✅ Handle Null Values

@Override
public byte[] serialize(String topic, User user) {
    if (user == null) {
        return null; // Kafka handles null records
    }
    // ... serialization logic
}

✅ Be Deterministic

// ❌ BAD: Non-deterministic (order matters!)
Map<String, String> map = new HashMap<>(); // Unordered!

// ✅ GOOD: Deterministic
Map<String, String> map = new TreeMap<>(); // Ordered

Why? Same object must always produce same bytes (for idempotence).

✅ Version Your Format

@Override
public byte[] serialize(String topic, User user) {
    ByteBuffer buffer = ByteBuffer.allocate(...);
    
    buffer.put((byte) 1); // Format version
    buffer.putLong(user.getId());
    // ... rest of data
    
    return buffer.array();
}

Why? Enables schema evolution and backward compatibility.

❌ Don't Include Business Logic

// ❌ BAD: Business logic in serializer
@Override
public byte[] serialize(String topic, User user) {
    if (user.getEmail() == null) {
        user.setEmail("noemail@example.com"); // Modifying object!
    }
    // ...
}

// ✅ GOOD: Validate before serialization
if (user.getEmail() == null) {
    throw new IllegalArgumentException("Email required");
}
producer.send(new ProducerRecord<>("users", user));

5. JSON Serialization

Using JsonSerializer (Kafka's Built-in)

Kafka provides a generic JSON serializer (Kafka 2.6+).

import org.apache.kafka.common.serialization.JsonSerializer;

props.put("value.serializer", 
    "org.apache.kafka.common.serialization.JsonSerializer");

// Send any POJO
User user = new User(123L, "Alice", "alice@example.com");
ProducerRecord<String, User> record = 
    new ProducerRecord<>("users", "123", user);
producer.send(record);

Result in Kafka:

{
  "id": 123,
  "name": "Alice",
  "email": "alice@example.com"
}

JSON with Jackson

import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.common.serialization.Serializer;

public class JacksonSerializer<T> implements Serializer<T> {
    private final ObjectMapper objectMapper = new ObjectMapper();
    
    @Override
    public byte[] serialize(String topic, T data) {
        if (data == null) {
            return null;
        }
        try {
            return objectMapper.writeValueAsBytes(data);
        } catch (Exception e) {
            throw new RuntimeException("Failed to serialize", e);
        }
    }
}

JSON Pros and Cons

✅ Pros:

• Human-readable (easy debugging)
• Language-agnostic
• Flexible (schema-less)
• Easy to get started

❌ Cons:

• Verbose (larger message size)
• No schema validation (errors caught late)
• Slower than binary formats
• No built-in versioning
• Type safety issues

Use When:

• Rapid prototyping
• Small message volumes
• Human readability is important
• Cross-language compatibility needed

6. Avro Serialization

What is Avro?

Apache Avro is a binary serialization format with:

• Schema definition (JSON format)
• Compact binary encoding
• Schema evolution support
• Strong typing

Avro Schema Example

{
  "type": "record",
  "name": "User",
  "namespace": "com.example",
  "fields": [
    {"name": "id", "type": "long"},
    {"name": "name", "type": "string"},
    {"name": "email", "type": "string"}
  ]
}

Using Avro with Kafka

import io.confluent.kafka.serializers.KafkaAvroSerializer;

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", 
    "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", 
    "io.confluent.kafka.serializers.KafkaAvroSerializer");
props.put("schema.registry.url", "http://localhost:8081");

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

// Create Avro record
Schema schema = new Schema.Parser().parse(schemaString);
GenericRecord user = new GenericData.Record(schema);
user.put("id", 123L);
user.put("name", "Alice");
user.put("email", "alice@example.com");

// Send
ProducerRecord<String, GenericRecord> record = 
    new ProducerRecord<>("users", "123", user);
producer.send(record);

Avro Pros and Cons

✅ Pros:

• Compact binary format (smaller messages)
• Schema enforcement (catch errors early)
• Schema evolution (backward/forward compatibility)
• Fast serialization/deserialization
• Code generation for strongly-typed objects

❌ Cons:

• Requires Schema Registry
• More setup complexity
• Not human-readable (debugging harder)
• Learning curve

Use When:

• Production systems with high volume
• Schema evolution is needed
• Performance matters
• Strong typing required

7. Protocol Buffers (Protobuf)

What is Protobuf?

Google's Protocol Buffers: binary serialization with schema.

Protobuf Schema (.proto file)

syntax = "proto3";

message User {
  int64 id = 1;
  string name = 2;
  string email = 3;
}

Compile Schema

protoc --java_out=. user.proto

Generates Java classes: User.java

Using Protobuf with Kafka

import com.google.protobuf.Message;
import org.apache.kafka.common.serialization.Serializer;

public class ProtobufSerializer<T extends Message> implements Serializer<T> {
    @Override
    public byte[] serialize(String topic, T data) {
        if (data == null) {
            return null;
        }
        return data.toByteArray();
    }
}

props.put("value.serializer", "com.myapp.ProtobufSerializer");

User user = User.newBuilder()
    .setId(123L)
    .setName("Alice")
    .setEmail("alice@example.com")
    .build();

ProducerRecord<String, User> record = 
    new ProducerRecord<>("users", "123", user);
producer.send(record);

Protobuf Pros and Cons

✅ Pros:

• Very compact
• Fast
• Strong typing
• Good language support (not just JVM)
• Schema evolution

❌ Cons:

• Code generation required
• Not human-readable
• Schema management needed

8. Key vs Value Serialization

Why Different Serializers?

// Key: Simple string (for partitioning)
// Value: Complex object (payload)
props.put("key.serializer", 
    "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", 
    "io.confluent.kafka.serializers.KafkaAvroSerializer");

Common Patterns:

Null Keys

// No key (null key)
ProducerRecord<String, String> record = 
    new ProducerRecord<>("topic", null, "value");

Effect: Round-robin partitioning (no ordering guarantee).

Partitioning details: Module 4.1

9. Schema Evolution

The Problem

// Version 1: User with id and name
class User {
    long id;
    String name;
}

// Version 2: Added email field
class User {
    long id;
    String name;
    String email; // NEW FIELD
}

Challenge: Old consumers can't read new format (or vice versa).

Evolution Strategies

Backward Compatibility

New consumers can read old data.

// V2 consumer reading V1 data
class User {
    long id;
    String name;
    String email = ""; // Default value for missing field
}

How to Achieve:

• Add fields with default values
• Don't remove fields
• Don't change field types

Forward Compatibility

Old consumers can read new data.

// V1 consumer reading V2 data (ignores email)
class User {
    long id;
    String name;
    // Ignores unknown "email" field
}

How to Achieve:

• Old code ignores unknown fields
• Don't change existing field semantics

Full Compatibility

Both backward and forward compatible.

Best achieved with:

• Avro (built-in schema evolution)
• Protobuf (field numbering)
• JSON Schema (with validation)

Avro Schema Evolution Example

Schema V1:

{
  "type": "record",
  "name": "User",
  "fields": [
    {"name": "id", "type": "long"},
    {"name": "name", "type": "string"}
  ]
}

Schema V2 (backward compatible):

{
  "type": "record",
  "name": "User",
  "fields": [
    {"name": "id", "type": "long"},
    {"name": "name", "type": "string"},
    {"name": "email", "type": "string", "default": ""}
  ]
}

Key: New field has a default value.

10. Common Serialization Failures

Failure 1: Mismatched Serializers

// Producer
props.put("value.serializer", "StringSerializer");
producer.send(new ProducerRecord<>("topic", "key", "value"));

// Consumer
props.put("value.deserializer", "IntegerDeserializer"); // ❌ Mismatch!

Error: SerializationException on consumer side.

Fix: Ensure producer and consumer use compatible serializers.

Failure 2: Null Handling

User user = null;
producer.send(new ProducerRecord<>("topic", "key", user));

What Happens:

• Serializer receives null
• Must return null (not throw exception)
• Kafka stores null value (tombstone in compacted topics)

Fix: Handle nulls in serializer:

if (data == null) return null;

Failure 3: Schema Changes Without Versioning

// Old format: "id,name"
// New format: "id,name,email"

Problem: Consumers expect old format.

Fix:

• Use schema versioning (Avro, Protobuf)
• Include version byte in custom serializers
• Use Schema Registry

Failure 4: Encoding Issues

// Producer uses UTF-8
String message = "Hello 世界";
byte[] bytes = message.getBytes(StandardCharsets.UTF_8);

// Consumer uses ISO-8859-1
String decoded = new String(bytes, StandardCharsets.ISO_8859_1); // ❌ Garbage

Fix: Always specify encoding explicitly.

Failure 5: Non-Deterministic Serialization

// ❌ BAD: HashMap order is non-deterministic
Map<String, String> map = new HashMap<>();
map.put("b", "2");
map.put("a", "1");
return objectMapper.writeValueAsBytes(map); // Order varies!

// ✅ GOOD: TreeMap has consistent order
Map<String, String> map = new TreeMap<>();

Why it matters: Idempotent producer compares byte arrays.

11. Best Practices

✅ Use Schema Registry (Production)

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

Benefits:

• Centralized schema management
• Automatic versioning
• Compatibility checks
• Schema evolution support

✅ Version Your Schemas

// Include version in custom formats
ByteBuffer buffer = ByteBuffer.allocate(size);
buffer.put((byte) 2); // Schema version
// ... rest of data

✅ Test Serialization Round-Trip

@Test
public void testSerializationRoundTrip() {
    User original = new User(123L, "Alice", "alice@example.com");
    
    // Serialize
    byte[] bytes = serializer.serialize("topic", original);
    
    // Deserialize
    User deserialized = deserializer.deserialize("topic", bytes);
    
    // Verify
    assertEquals(original, deserialized);
}

✅ Handle Backward Compatibility

// New field with default
String email = user.get("email") != null 
    ? user.get("email") 
    : "noemail@example.com";

✅ Document Your Schema

/**
 * User serialization format (Version 2)
 * 
 * Byte layout:
 *   0: Version byte (0x02)
 *   1-8: User ID (long)
 *   9-12: Name length (int)
 *   13-...: Name bytes (UTF-8)
 *   ...: Email length (int)
 *   ...: Email bytes (UTF-8)
 */

❌ Avoid Serializing Transient State

class User {
    long id;
    String name;
    transient String sessionToken; // ❌ Don't serialize
}

12. Choosing a Serialization Format

Decision Matrix

Recommendation Flow

START
  │
  ├─ Prototyping? → JSON
  │
  ├─ High volume production? → Avro or Protobuf
  │
  ├─ Cross-language? → Protobuf
  │
  ├─ Schema evolution critical? → Avro
  │
  ├─ Simple text data? → String
  │
  └─ Special requirements? → Custom Binary

Summary

Key Takeaways

✅ Serialization converts objects to bytes (and back)
✅ Producer serializes, consumer deserializes (must match)
✅ Built-in serializers for common types (String, Integer, byte[])
✅ Custom serializers for domain objects
✅ JSON for readability and flexibility
✅ Avro/Protobuf for performance and schema evolution
✅ Schema Registry for production systems
✅ Version your schemas for compatibility
✅ Test round-trip serialization

Serialization Checklist

Before going to production:

• Serializers are deterministic
• Null handling implemented
• Schema versioning in place
• Backward compatibility tested
• Round-trip tests written
• Schema Registry configured (if using Avro)
• Documentation updated with format details
• Performance tested with realistic data

What's Next

• Module 4.1: Partitioning Strategy & Message Ordering
  Learn how keys affect partitioning and ordering

• Module 5: Reliability, Delivery Guarantees & Semantics
  Ensure your serialized data is delivered reliably

Previous Module: ← 2.1 - Producer APIs & Core Configuration
Next Module: 4.1 - Partitioning Strategy & Message Ordering →