Serialization #

In the Apache Kafka distributed system, data is transmitted and stored as raw bytes. Kafka brokers have no understanding of the object structures, classes, or data schemas we use at the application level. The broker only receives byte arrays from producers, stores them in commit log files on disk, and sends them back to consumers. Therefore, the responsibility of converting application memory objects into byte arrays before sending rests entirely with the producer through a process called Serialization. Conversely, converting those byte arrays back into application objects on the receiving side is called Deserialization. This design gives us unlimited flexibility to use any programming language, but on the other hand, it demands very strict discipline in managing data structures to avoid system failures in production.

The Importance of Serialization and the Serializer’s Role #

The serialization process is the second step in the message delivery lifecycle after the producer ensures cluster metadata is available in the local memory cache. Every time we create a ProducerRecord<K, V> object, we must specify the serializer class for the message key and value.

The Kafka producer client (Java) uses two main configuration properties to specify these serializers:

key.serializer=org.apache.kafka.common.serialization.StringSerializer
value.serializer=org.apache.kafka.common.serialization.StringSerializer

Kafka provides several built-in serializers covering basic data types:

• StringSerializer: Converts Java strings to bytes using UTF-8 encoding.
• IntegerSerializer / LongSerializer / DoubleSerializer: Convert numbers to fixed binary representations (for example, 8 bytes for Long).
• ByteArraySerializer: Skips the conversion process because the data is already raw bytes (byte[]). This is useful if our application does serialization manually before calling Kafka.

Although these built-in serializers are very useful for simple applications or plain string message delivery, enterprise-scale systems need far more complex data structures, like user profile objects, payment transactions, or IoT device activity logs. This is where the decision to choose an object serialization format becomes a crucial factor determining network efficiency, disk storage usage, and integration stability between microservices.

The Dangers of Custom or Plain JSON Serialization #

When faced with the need to send complex data objects, developers often take shortcuts that seem easy but hide big risks down the road. There are two popular approaches categorized as bad practices (anti-patterns):

1. Java Native Object Serialization #

Implementing the java.io.Serializable interface and using ObjectOutputStream to produce byte arrays.

• Critical Weakness: This approach tightly couples our application entirely to the Java platform. Consumer services written in Go, Python, C#, or Node.js won’t be able to easily deserialize those messages. Additionally, Java serialization is very sensitive to class version changes (serialVersionUID), has huge byte size overhead because it includes long class names, and is vulnerable to deserialization vulnerabilities.

2. Converting Objects to JSON Strings (Vanilla JSON Serialization) #

Using JSON parsers like Gson, Jackson, or similar libraries to convert objects to plain JSON text strings, then sending them using StringSerializer.

• Critical Weakness: JSON is a text-based format that wastes a lot of space. Every JSON message must include property key names repeatedly (for example, the strings "transaction_id" and "customer_email" are sent in every message). When our application throughput reaches millions of messages per second, this text metadata overhead wastes network bandwidth and cluster disk storage for nothing.
• No Formal Schema: JSON has no data contract guarantee. Producers can easily add, remove, or change a property’s data type suddenly without any producer-side validation. As a result, consumers expecting certain properties crash at runtime due to mismatched data types.

Here’s a code illustration comparing plain JSON serialization handling (anti-pattern) with safe schema-based serialization:

// ANTI-PATTERN: Manual JSON serialization that wastes bandwidth and is unsafe
public class OrderService {
    public void sendOrder(KafkaProducer<String, String> producer, Order order) {
        // ✗ JSON strings carry field names like "orderId", "customerId" in every message
        String jsonPayload = String.format(
            "{\"orderId\":\"%s\",\"customerId\":\"%s\",\"amount\":%.2f}",
            order.getId(), order.getCustomerId(), order.getAmount()
        );
        
        ProducerRecord<String, String> record = new ProducerRecord<>("orders", order.getId(), jsonPayload);
        producer.send(record);
    }
}

// CORRECT: Binary schema-based serialization (e.g., Apache Avro) with Schema Registry
// The schema is validated at compile time, the payload is pure binary without repeating text field names
public class OrderServiceSecure {
    public void sendOrder(KafkaProducer<String, OrderEvent> producer, OrderEvent order) {
        // ✓ The 'OrderEvent' object is a Java class auto-generated from an official Avro schema (.avsc)
        ProducerRecord<String, OrderEvent> record = new ProducerRecord<>("orders-avro", order.getOrderId(), order);
        producer.send(record);
    }
}

Schema Evolution and the Schema Drift Problem #

In event-based microservices architectures, producers and consumers are developed by different teams, using different code repositories, and deployed at different times. Inevitable business requirement changes force us to modify data structures. This phenomenon triggers a problem known as Schema Drift — the condition where the data format sent by producers is no longer in sync with consumer expectations.

If we don’t have a systematic schema evolution handling mechanism, small changes like removing a column or changing a data type from Integer to Double can collapse our entire downstream pipelines.

To manage schema changes safely, we must agree on one of the following three compatibility rules:

Golden Rules of Schema Evolution: #

1. Always define default values for every new field we add. This ensures backward compatibility because new consumers can use those defaults when reading old messages that lack the field.
2. Never delete required fields. If a field must be removed, first change its nature to optional in an intermediate version.
3. Never drastically change the data type of existing fields (for example, from String to Array).

Getting to Know the Confluent Schema Registry #

To apply schema compatibility rules consistently and automatically, we need a central authority called the Confluent Schema Registry. The Schema Registry is an external service running outside the Kafka cluster. It acts as a central repository for storing, managing, and validating the data schemas used in the Kafka cluster.

How Does the Schema Registry Reduce Payload Size? #

Instead of including the entire schema structure (which can be several KB) in every message sent to Kafka, the Schema Registry only assigns a unique integer ID (4 bytes) to each registered schema.

When a producer sends data using a Schema Registry serializer (like KafkaAvroSerializer), the serializer:

1. Checks whether the object’s schema is already registered in the Schema Registry.
2. If not, the schema is registered and the Schema Registry returns a unique Schema ID (for example, ID 42).
3. The serializer creates a binary payload with a special format:
   • Byte 0: Magic Byte (always 0 to mark the Schema Registry format).
   • Bytes 1-4: The Schema ID as a 4-byte integer (for example, the binary representation of 42).
   • Bytes 5+: The actual serialized binary data payload without text field names.
4. The total added overhead is only 5 bytes! This is far more efficient than JSON.

When a consumer receives a message, the deserializer reads the Schema ID from the first 5 bytes, downloads the matching schema from the Schema Registry (if not already in the consumer’s local cache), then deserializes the remaining payload bytes with high precision.

Interaction Workflow with the Schema Registry #

Here’s the asynchronous interaction flow between the producer, consumer, Schema Registry, and Kafka broker for validating and transmitting data:

flowchart TD
    subgraph Client["Producer Client"]
        App["Application Thread"]
        KafkaProd["Kafka Producer Client"]
        LocalCache["Local Schema Cache"]
    end
    
    subgraph Registry["Confluent Schema Registry"]
        SR["Schema Registry Server"]
        SchemaDB["Schema Storage (_schemas topic)"]
    end
    
    subgraph BrokerCluster["Apache Kafka Cluster"]
        KB["Kafka Broker (orders topic)"]
    end
    
    App -->|"1. Send Avro Object"| KafkaProd
    KafkaProd -->|"2. Check Local Cache for Schema"| LocalCache
    
    LocalCache -. "3a. Hit: Use Schema ID" .-> SendBroker
    LocalCache -. "3b. Miss: Register Schema" .-> SR
    
    SR -->|"4. Validate Schema Compatibility"| SchemaDB
    SchemaDB -->|"5. Provide New/Registered Schema ID"| SR
    SR -. "6. Return Schema ID (e.g., ID 42)" .-> KafkaProd
    
    KafkaProd -->|"7. Store ID 42 in Local Cache"| LocalCache
    KafkaProd -->|"8. Serialize Payload (Add Schema ID in Magic Byte)"| SendBroker["9. Send Byte Payload (Magic Byte + ID 42 + Data)"]
    
    SendBroker --> KB
    
    style Client stroke:#e5e7eb
    style Registry stroke:#e5e7eb
    style BrokerCluster stroke:#e5e7eb
    style LocalCache stroke:#0288d1,stroke-width:2px
    style SR stroke:#2e7d32,stroke-width:2px

Modern Serialization Formats: Avro vs Protobuf #

The two most popular binary serialization formats fully supported by the Kafka ecosystem and Schema Registry are Apache Avro and Protocol Buffers (Protobuf).

1. Apache Avro #

Avro is a row-oriented serialization format developed in the Apache Hadoop project. Avro’s main characteristics:

• JSON Schema: Avro schemas are defined using JSON-format files with the .avsc extension.
• Binary Encoding: Data is serialized into a very compact binary format. Without the correct reader schema, raw Avro bytes can’t be parsed at all.
• Dynamic Typing: Avro doesn’t require code generation, although in production it’s recommended to use generator plugins to produce type-safe Java classes.

2. Protocol Buffers (Protobuf) #

Protobuf is a binary serialization format developed by Google. Protobuf’s main characteristics:

• .proto Schema: Schemas are defined using special syntax in .proto files.
• Mandatory Code Generation: Protobuf heavily relies on the protoc compiler to generate object-builder classes in various programming languages.
• Tag Numbers: Every field in Protobuf is identified by a unique integer tag number (for example string name = 1;). This makes schema evolution very efficient because renaming fields doesn’t affect binary compatibility as long as tag numbers stay the same.

Serialization Format Comparison Table #

Real Avro Implementation with Schema Registry #

Let’s build a real implementation example using Apache Avro in a Java environment.

Step 1: Defining the Avro Schema (OrderEvent.avsc) #

Save this file in the src/main/avro/OrderEvent.avsc directory:

{
  "type": "record",
  "name": "OrderEvent",
  "namespace": "com.unisbadri.kafka.model",
  "doc": "Schema representing customer order transactions.",
  "fields": [
    {
      "name": "orderId",
      "type": "string",
      "doc": "Unique order transaction ID."
    },
    {
      "name": "customerId",
      "type": "string",
      "doc": "The customer ID who made the transaction."
    },
    {
      "name": "totalAmount",
      "type": "double",
      "doc": "Total transaction value in Rupiah."
    },
    {
      "name": "status",
      "type": "string",
      "default": "CREATED",
      "doc": "Current transaction status (e.g., CREATED, PAID, SHIPPED)."
    }
  ]
}

Step 2: Avro Producer Java Code #

After running the Maven/Gradle compilation to generate the OrderEvent Java class, we can use it in our producer code:

import io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig;
import io.confluent.kafka.serializers.KafkaAvroSerializer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import com.unisbadri.kafka.model.OrderEvent;
import java.util.Properties;

public class AvroProducerApp {
    public static void main(String[] args) {
        Properties props = new Properties();
        
        // 1. Basic broker connection configuration
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.CLIENT_ID_CONFIG, "AvroProducerClient");
        
        // 2. Specify serializers for key and value
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        
        // ✓ CORRECT: Using Confluent's built-in KafkaAvroSerializer for value serialization
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class.getName());
        
        // 3. Schema Registry URL configuration
        // ✓ CORRECT: Pointing the serializer to the central Schema Registry server
        props.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081");
        
        // 4. Additional security configuration (Optional locally, mandatory in staging/production)
        // props.put("basic.auth.credentials.source", "USER_INFO");
        // props.put("basic.auth.user.info", "apiKey:***");

        try (KafkaProducer<String, OrderEvent> producer = new KafkaProducer<>(props)) {
            // Creating a data object using the Avro builder pattern
            OrderEvent order = OrderEvent.newBuilder()
                .setOrderId("ORD-2026-0001")
                .setCustomerId("CUST-1002")
                .setTotalAmount(1250000.0)
                .setStatus("CREATED")
                .build();

            ProducerRecord<String, OrderEvent> record = new ProducerRecord<>(
                "orders-avro-topic", 
                order.getOrderId().toString(), 
                order
            );

            System.out.println("Sending Avro transaction to Kafka...");
            producer.send(record, (metadata, exception) -> {
                if (exception != null) {
                    // DON'T: ignore network errors or schema rejections
                    System.err.println("Failed to send data due to schema violation or error: " + exception.getMessage());
                } else {
                    // ✓ Successfully wrote binary data
                    System.out.printf("Success! Data stored on partition %d, offset %d%n", 
                        metadata.partition(), metadata.offset());
                }
            });
            
            // Call flush to ensure data leaves the local buffer pool
            producer.flush();
        } catch (Exception e) {
            System.err.println("Fatal error in Avro producer: " + e.getMessage());
        }
    }
}

Detecting and Preventing Schema Violations in Production #

The best prevention against schema drift problems doesn’t happen at application runtime in production, but before our application code is deployed to servers. We must integrate schema compatibility checks directly into our CI/CD pipeline.

1. Schema Validation Integration in the CI/CD Pipeline #

Confluent provides Maven and Gradle plugins for testing schema compatibility against an active Schema Registry server before running the jar/package build process.

Example Maven plugin integration (pom.xml):

<plugin>
    <groupId>io.confluent</groupId>
    <artifactId>kafka-schema-registry-maven-plugin</artifactId>
    <version>7.5.0</version>
    <configuration>
        <schemaRegistryUrls>
            <schemaRegistryUrl>http://localhost:8081</schemaRegistryUrl>
        </schemaRegistryUrls>
        <subjects>
            <!-- Associating the topic name with the schema class name -->
            <orders-avro-topic-value>src/main/avro/OrderEvent.avsc</orders-avro-topic-value>
        </subjects>
    </configuration>
    <executions>
        <execution>
            <id>test-compatibility</id>
            <phase>test</phase>
            <goals>
                <!-- Running the compatibility test command against active schemas in the registry -->
                <goal>test-compatibility</goal>
            </goals>
        </execution>
    </executions>
</plugin>

When developers release new code with modified .avsc files, the following command runs in the CI/CD runner:

# Schema compatibility check command
mvn schema-registry:test-compatibility

If the schema change is deemed to violate compatibility rules (for example, deleting a required field without a default value under backward compatibility), the build process fails automatically, preventing the problematic code from reaching our production servers.

2. Handling Corrupted Deserialization Records (Poison Pill) #

On the consumer side, if a message fails to deserialize due to corrupted byte format or an invalid schema ID, the consumption process can get stuck. This damaging message is called a Poison Pill.

• Handling Strategy: We must not let the consumer die repeatedly. We must use a special deserializer like Spring Kafka’s ErrorHandlingDeserializer that catches deserialization errors, logs the exception, and routes the corrupted message to a dedicated Dead Letter Queue (DLQ) topic for manual investigation by the engineering team.

Summary #

• Pure Binary: Apache Kafka stores data as raw byte arrays without caring about our application’s original object format structure.
• JSON Danger: Using plain text JSON for high-throughput data triggers network bandwidth waste because field names are repeated in every message.
• Schema Drift: The data structure desynchronization problem between producers and consumers that can trigger runtime crashes in downstream applications.
• Schema Registry: A centralized service for storing schema contracts and assigning unique 4-byte integer IDs to reduce transmission payload byte size.
• Compatibility: Schema evolution rules (Backward, Forward, Full) governing how schemas are modified without breaking old consumer applications.
• CI/CD Validation: Use the Schema Registry plugin in maven/gradle to test schema modification feasibility before code is deployed to production.

← Previous: Producer Workflow Next: Key vs No-Key Message →