Custom Serializer
Kafka stores and transports byte arrays in its topics. If you wish to use the Java Objects instead of byte arrays, Kafka provides interfaces to accomplish it.
To implement a serializer, create a class that implements the org.apache.kafka.common.serialization.Serializer interface.
To implement a deserializer, create a class that implements the org.apache.kafka.common.serialization.Deserializer interface.
To implement serializer and deserializer, you need to implement the following methods:
configure()
close()
serialize() / deserialize()
Serializing MyMessage in Producer
You create a serializer class that implements org.apache.kafka.common.serialization.Serializer.
serialize() receives your object and returns a serialized version as bytes array.
For example, Serializing MyMessage object in Producer is as follows:
public class MyValueSerializer implements Serializer<MyMessage>
{
private boolean isKey;
@Override
public void configure(Map<String, ?> configs, boolean isKey)
{
this.isKey = isKey;
}
@Override
public byte[] serialize(String topic, MyMessage message)
{
if (message == null) {
return null;
}
try {
(serialize your MyMessage object into bytes)
return bytes;
} catch (IOException | RuntimeException e) {
throw new SerializationException("Error serializing value", e);
}
}
@Override
public void close()
{
}
}
Deserializing MyMessage in Consumer
You create a deserializer class that implements org.apache.kafka.common.serialization.Deserializer
deserialize() receives serialized value as bytes array and returns your object.
For example, Deserializing MyMessage object in Consumer is as follows:
public class MyValueDeserializer implements Deserializer<MyMessage>
{
private boolean isKey;
@Override
public void configure(Map<String, ?> configs, boolean isKey)
{
this.isKey = isKey;
}
@Override
public MyMessage deserialize(String s, byte[] value)
{
if (value == null) {
return null;
}
try {
(deserialize value into your MyMessage object)
MyMessage message = new MyMessage();
return message;
} catch (IOException | RuntimeException e) {
throw new SerializationException("Error deserializing value", e);
}
}
@Override
public void close()
{
}
}
Note:
Like Kafka message value, Kafka message key also follows the same process in serializing and deserializing the objects.