How to Implement Custom Serializers and Deserializers in Apache Kafka?

Apache Kafka uses byte arrays to transmit records across topics efficiently. When working with custom domain objects in Java, Kafka needs a mechanism to convert those objects into byte streams before sending them to a topic (serialization) and back into Java objects upon receiving them (deserialization).

While Kafka provides built-in serializers for standard data types like String, Integer, and Long, real-world applications often require processing custom Java POJOs. This guide demonstrates how to implement custom serializers and deserializers using Jackson JSON library.

1. Define the Domain Model

First, create a simple Java object (POJO) that will be transferred over Kafka topics.

package com.example.kafka;

public class User {
    private String userId;
    private String name;
    private String email;

    public User() {}

    public User(String userId, String name, String email) {
        this.userId = userId;
        this.name = name;
        this.email = email;
    }

    public String getUserId() { return userId; }
    public void setUserId(String userId) { this.userId = userId; }
    public String getName() { return name; }
    public void setName(String name) { this.name = name; }
    public String getEmail() { return email; }
    public void setEmail(String email) { this.email = email; }
}

2. Implement the Custom Serializer

To create a custom serializer, implement Kafka's Serializer<T> interface and override the serialize method.

package com.example.kafka;

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

public class UserSerializer implements Serializer<User> {
    private final ObjectMapper objectMapper = new ObjectMapper();

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        // Initialization logic if required
    }

    @Override
    public byte[] serialize(String topic, User data) {
        if (data == null) {
            return null;
        }
        try {
            return objectMapper.writeValueAsBytes(data);
        } catch (Exception e) {
            throw new RuntimeException("Error serializing User object", e);
        }
    }

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

3. Implement the Custom Deserializer

To convert incoming byte arrays back into your domain object, implement Kafka's Deserializer<T> interface.

package com.example.kafka;

import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.common.serialization.Deserializer;
import java.util.Map;

public class UserDeserializer implements Deserializer<User> {
    private final ObjectMapper objectMapper = new ObjectMapper();

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        // Initialization logic if required
    }

    @Override
    public User deserialize(String topic, byte[] data) {
        if (data == null || data.length == 0) {
            return null;
        }
        try {
            return objectMapper.readValue(data, User.class);
        } catch (Exception e) {
            throw new RuntimeException("Error deserializing byte array to User object", e);
        }
    }

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

4. Configure Producer and Consumer Properties

Once your serializer and deserializer classes are created, register them in your Kafka client properties using value.serializer and value.deserializer properties.

Producer Configuration

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "com.example.kafka.UserSerializer");

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

Consumer Configuration

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "user-group");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "com.example.kafka.UserDeserializer");

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

Best Practices

• Error Handling: Catch and log transformation errors appropriately in serialize and impulse methods to prevent silent failures or application crashes.
• Schema Evolution: For complex, production-grade applications, consider using Apache Avro with Confluent Schema Registry instead of pure custom JSON SerDes to safely handle schema changes over time.
• Stateless SerDes: Keep your serializer and deserializer implementations stateless when possible to facilitate thread safety and instance reuse.