Kafka Serialization¶

Kafka transmits messages as byte arrays. Serializers convert objects to bytes for producers, and deserializers convert bytes back to objects for consumers. This guide covers serialization formats, Schema Registry integration, and best practices.

Serialization Overview¶

Data Flow¶

Serialization Formats¶

Format	Schema	Type Safety	Size	Human Readable
String	None	None	Large	Yes
JSON	Optional	Runtime	Large	Yes
Avro	Required	Compile-time	Compact	No
Protobuf	Required	Compile-time	Compact	No
Byte Array	None	None	Minimal	No

Built-in Serializers¶

StringSerializer¶

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

Producer<String, String> producer = new KafkaProducer<>(props);
producer.send(new ProducerRecord<>("topic", "key", "value"));

ByteArraySerializer¶

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

Producer<byte[], byte[]> producer = new KafkaProducer<>(props);
producer.send(new ProducerRecord<>("topic", keyBytes, valueBytes));

Available Serializers¶

Type	Serializer	Deserializer
String	`StringSerializer`	`StringDeserializer`
Byte Array	`ByteArraySerializer`	`ByteArrayDeserializer`
Integer	`IntegerSerializer`	`IntegerDeserializer`
Long	`LongSerializer`	`LongDeserializer`
Double	`DoubleSerializer`	`DoubleDeserializer`
Float	`FloatSerializer`	`FloatDeserializer`
Short	`ShortSerializer`	`ShortDeserializer`
UUID	`UUIDSerializer`	`UUIDDeserializer`
Void	`VoidSerializer`	`VoidDeserializer`

JSON Serialization¶

Jackson Serializer¶

public class JsonSerializer<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 (JsonProcessingException e) {
            throw new SerializationException("Error serializing JSON message", e);
        }
    }
}

public class JsonDeserializer<T> implements Deserializer<T> {
    private final ObjectMapper objectMapper = new ObjectMapper();
    private Class<T> targetType;

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        targetType = (Class<T>) configs.get("json.value.type");
    }

    @Override
    public T deserialize(String topic, byte[] data) {
        if (data == null) return null;

        try {
            return objectMapper.readValue(data, targetType);
        } catch (IOException e) {
            throw new SerializationException("Error deserializing JSON message", e);
        }
    }
}

Usage¶

// Producer
props.put("value.serializer", JsonSerializer.class);

// Consumer
props.put("value.deserializer", JsonDeserializer.class);
props.put("json.value.type", Order.class);

Spring Kafka JSON Support¶

@Configuration
public class KafkaConfig {

    @Bean
    public ProducerFactory<String, Order> producerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        return new DefaultKafkaProducerFactory<>(config);
    }

    @Bean
    public ConsumerFactory<String, Order> consumerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        config.put(JsonDeserializer.VALUE_DEFAULT_TYPE, Order.class);
        config.put(JsonDeserializer.TRUSTED_PACKAGES, "com.example.domain");
        return new DefaultKafkaConsumerFactory<>(config);
    }
}

Avro Serialization¶

Schema Definition¶

{
  "type": "record",
  "name": "Order",
  "namespace": "com.example.avro",
  "fields": [
    {"name": "orderId", "type": "string"},
    {"name": "customerId", "type": "string"},
    {"name": "amount", "type": "double"},
    {"name": "currency", "type": "string", "default": "USD"},
    {"name": "createdAt", "type": {"type": "long", "logicalType": "timestamp-millis"}},
    {
      "name": "items",
      "type": {
        "type": "array",
        "items": {
          "type": "record",
          "name": "OrderItem",
          "fields": [
            {"name": "productId", "type": "string"},
            {"name": "quantity", "type": "int"},
            {"name": "price", "type": "double"}
          ]
        }
      }
    }
  ]
}

Avro with Schema Registry¶

Properties props = new Properties();
props.put("bootstrap.servers", "kafka:9092");
props.put("schema.registry.url", "http://schema-registry:8081");
props.put("key.serializer", StringSerializer.class);
props.put("value.serializer", KafkaAvroSerializer.class);

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

Order order = Order.newBuilder()
    .setOrderId("order-123")
    .setCustomerId("customer-456")
    .setAmount(99.99)
    .setCurrency("USD")
    .setCreatedAt(System.currentTimeMillis())
    .setItems(List.of(
        OrderItem.newBuilder()
            .setProductId("product-789")
            .setQuantity(2)
            .setPrice(49.99)
            .build()
    ))
    .build();

producer.send(new ProducerRecord<>("orders", order.getOrderId(), order));

Consumer with Avro¶

Properties props = new Properties();
props.put("bootstrap.servers", "kafka:9092");
props.put("schema.registry.url", "http://schema-registry:8081");
props.put("group.id", "order-consumers");
props.put("key.deserializer", StringDeserializer.class);
props.put("value.deserializer", KafkaAvroDeserializer.class);
props.put("specific.avro.reader", "true");  // Use generated classes

Consumer<String, Order> consumer = new KafkaConsumer<>(props);

Generic Avro Records¶

For dynamic schemas:

props.put("specific.avro.reader", "false");  // Use GenericRecord

Consumer<String, GenericRecord> consumer = new KafkaConsumer<>(props);
consumer.subscribe(List.of("orders"));

while (running) {
    ConsumerRecords<String, GenericRecord> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, GenericRecord> record : records) {
        GenericRecord order = record.value();
        String orderId = order.get("orderId").toString();
        double amount = (double) order.get("amount");
    }
}

Protobuf Serialization¶

Proto Definition¶

syntax = "proto3";

package com.example.proto;

option java_package = "com.example.proto";
option java_outer_classname = "OrderProtos";

message Order {
  string order_id = 1;
  string customer_id = 2;
  double amount = 3;
  string currency = 4;
  int64 created_at = 5;
  repeated OrderItem items = 6;
}

message OrderItem {
  string product_id = 1;
  int32 quantity = 2;
  double price = 3;
}

Protobuf with Schema Registry¶

Properties props = new Properties();
props.put("bootstrap.servers", "kafka:9092");
props.put("schema.registry.url", "http://schema-registry:8081");
props.put("key.serializer", StringSerializer.class);
props.put("value.serializer", KafkaProtobufSerializer.class);

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

Order order = Order.newBuilder()
    .setOrderId("order-123")
    .setCustomerId("customer-456")
    .setAmount(99.99)
    .setCurrency("USD")
    .setCreatedAt(System.currentTimeMillis())
    .addItems(OrderItem.newBuilder()
        .setProductId("product-789")
        .setQuantity(2)
        .setPrice(49.99)
        .build())
    .build();

producer.send(new ProducerRecord<>("orders", order.getOrderId(), order));

Schema Registry Integration¶

Schema Registry Architecture¶

Configuration¶

// Producer
props.put("schema.registry.url", "http://schema-registry:8081");
props.put("auto.register.schemas", "true");  // Automatic schema registration
props.put("value.subject.name.strategy",
    "io.confluent.kafka.serializers.subject.TopicRecordNameStrategy");

// Consumer
props.put("schema.registry.url", "http://schema-registry:8081");
props.put("specific.avro.reader", "true");

Subject Naming Strategies¶

Strategy	Subject Name	Use Case
`TopicNameStrategy`	`{topic}-value`	One schema per topic
`RecordNameStrategy`	`{record.name}`	Schema per record type
`TopicRecordNameStrategy`	`{topic}-{record.name}`	Multiple schemas per topic

Schema Evolution¶

Compatibility Types¶

Type	Allowed Changes
BACKWARD	Delete fields, add optional fields
FORWARD	Add fields, delete optional fields
FULL	Add optional fields, delete optional fields
NONE	Any change allowed

Backward Compatible Evolution¶

// Version 1
{
  "type": "record",
  "name": "Order",
  "fields": [
    {"name": "orderId", "type": "string"},
    {"name": "amount", "type": "double"}
  ]
}

// Version 2 - Add optional field (BACKWARD compatible)
{
  "type": "record",
  "name": "Order",
  "fields": [
    {"name": "orderId", "type": "string"},
    {"name": "amount", "type": "double"},
    {"name": "currency", "type": "string", "default": "USD"}
  ]
}

Forward Compatible Evolution¶

// Version 1
{
  "type": "record",
  "name": "Order",
  "fields": [
    {"name": "orderId", "type": "string"},
    {"name": "amount", "type": "double"},
    {"name": "notes", "type": ["null", "string"], "default": null}
  ]
}

// Version 2 - Remove optional field (FORWARD compatible)
{
  "type": "record",
  "name": "Order",
  "fields": [
    {"name": "orderId", "type": "string"},
    {"name": "amount", "type": "double"}
  ]
}

Setting Compatibility¶

# Set compatibility for a subject
curl -X PUT -H "Content-Type: application/vnd.schemaregistry.v1+json" \
    --data '{"compatibility": "BACKWARD"}' \
    http://schema-registry:8081/config/orders-value

Custom Serializers¶

Implementing a Custom Serializer¶

public class OrderSerializer implements Serializer<Order> {
    private final ObjectMapper mapper = new ObjectMapper();
    private boolean isKey;

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        this.isKey = isKey;
        // Apply any configuration
    }

    @Override
    public byte[] serialize(String topic, Order order) {
        if (order == null) return null;

        try {
            return mapper.writeValueAsBytes(order);
        } catch (JsonProcessingException e) {
            throw new SerializationException("Error serializing Order", e);
        }
    }

    @Override
    public byte[] serialize(String topic, Headers headers, Order order) {
        // Add metadata to headers if needed
        headers.add("serializer.version", "1".getBytes());
        return serialize(topic, order);
    }

    @Override
    public void close() {
        // Cleanup resources if needed
    }
}

Topic-Specific Serialization¶

public class DelegatingSerializer implements Serializer<Object> {
    private final Map<String, Serializer<?>> serializers = new HashMap<>();

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        // Configure serializers per topic
        serializers.put("orders", new OrderSerializer());
        serializers.put("payments", new PaymentSerializer());
        serializers.put("events", new JsonSerializer<>());
    }

    @Override
    public byte[] serialize(String topic, Object data) {
        Serializer serializer = serializers.getOrDefault(topic, new JsonSerializer<>());
        return serializer.serialize(topic, data);
    }
}

Error Handling¶

Deserialization Errors¶

// Use ErrorHandlingDeserializer to catch errors
props.put("value.deserializer", ErrorHandlingDeserializer.class);
props.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, JsonDeserializer.class);
props.put(ErrorHandlingDeserializer.VALUE_FUNCTION, MyFailedDeserializationHandler.class);

public class MyFailedDeserializationHandler implements Function<FailedDeserializationInfo, Order> {
    @Override
    public Order apply(FailedDeserializationInfo info) {
        log.error("Failed to deserialize message from topic {}: {}",
            info.getTopic(), info.getException().getMessage());
        // Return null or a default value
        return null;
    }
}

Schema Mismatch Handling¶

try {
    Order order = consumer.poll(Duration.ofMillis(100)).iterator().next().value();
} catch (SerializationException e) {
    if (e.getCause() instanceof SchemaException) {
        // Schema incompatibility
        log.error("Schema mismatch: {}", e.getMessage());
        // Update consumer or handle gracefully
    }
}

Performance Considerations¶

Serialization Overhead¶

Format	Serialization Speed	Message Size	CPU Usage
Byte Array	Fastest	Smallest	Lowest
Protobuf	Fast	Small	Low
Avro	Fast	Small	Low
JSON	Slow	Large	High

Optimization Tips¶

// Reuse ObjectMapper (thread-safe)
private static final ObjectMapper MAPPER = new ObjectMapper();

// Use byte array buffer pool
props.put("buffer.memory", 67108864);  // 64MB

// Enable compression for large messages
props.put("compression.type", "lz4");

Best Practices¶

Format Selection¶

Scenario	Recommended Format
Strict schema enforcement	Avro or Protobuf
Cross-language compatibility	Avro or Protobuf
Debugging / human readable	JSON
Maximum performance	Protobuf or custom
Flexible schema	JSON

Schema Management¶

Practice	Recommendation
Version schemas	Use Schema Registry
Enforce compatibility	Set compatibility mode
Test evolution	Validate changes before deployment
Document schemas	Include field descriptions

Error Handling¶

Practice	Recommendation
Handle null	Check for null before serialization
Catch errors	Use ErrorHandlingDeserializer
Log failures	Include topic, partition, offset
DLQ	Send undeserializable messages to DLQ

Schema Registry - Schema management
Schema Formats - Format comparison
Producer Development - Producer patterns
Consumer Development - Consumer patterns

Kafka Serialization¶

Serialization Overview¶

Data Flow¶

Serialization Formats¶

Built-in Serializers¶

StringSerializer¶

ByteArraySerializer¶

Available Serializers¶

JSON Serialization¶

Jackson Serializer¶

Usage¶

Spring Kafka JSON Support¶

Avro Serialization¶

Schema Definition¶

Avro with Schema Registry¶

Consumer with Avro¶

Generic Avro Records¶

Protobuf Serialization¶

Proto Definition¶

Protobuf with Schema Registry¶

Schema Registry Integration¶

Schema Registry Architecture¶

Configuration¶

Subject Naming Strategies¶

Schema Evolution¶

Compatibility Types¶

Backward Compatible Evolution¶

Forward Compatible Evolution¶

Setting Compatibility¶

Custom Serializers¶

Implementing a Custom Serializer¶

Topic-Specific Serialization¶

Error Handling¶

Deserialization Errors¶

Schema Mismatch Handling¶

Performance Considerations¶

Serialization Overhead¶

Optimization Tips¶

Best Practices¶

Format Selection¶

Schema Management¶

Error Handling¶

Related Documentation¶