How to Implement Custom Serializers and Deserializers in Apache Kafka?

When transmitting complex POJOs or custom objects over Kafka topics, custom serializers and deserializers (SerDes) allow you to convert Java objects into byte arrays and vice versa.

Implementing Custom Serializer

Implement the org.apache.kafka.common.serialization.Serializer interface and override the serialize method.

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

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

Implementing Custom Deserializer

Implement the org.apache.kafka.common.serialization.Deserializer interface to convert raw byte arrays back into domain models.

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

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

After implementing these classes, set them in your producer and consumer properties under key.serializer or value.serializer.