Page History

...

• New class org.apache.kafka.common.serialization.ListSerializer which implements the Serializer<List<T>> Serializer<List<T>> interface
• New class org.apache.kafka.common.serialization.ListDeserializer which implements Deserializer<List<T>> Deserializer<List<T>> interface
• New subclass ListSerde<T> in org.apache.kafka.common.serialization.Serdes which creates new serde based on ListSerializer and ListDeserializer classes
• New method public static <T> Serde<List<T>> ListSerde() in org.apache.kafka.common.serialization.Serdes class (infers list implementation and inner serde from config file)
• New method public static <T> Serde<List<T>> ListSerde(Class listClass, Serde<T> innerSerde) in org.apache.kafka.common.serialization.Serdes class

...

First, as usually we need to specify our list serde using:

default.key/value.serde serde = org.apache.kafka.common.serialization.Serdes$ListSerde

Then, we need to cover those two configurations and here I'm proposing these two extra properties:

default.list.key/value.list.serde.impl type = java.util.ArrayList

default.list.key/value.list.serde.element inner = org.apache.kafka.common.serialization.Serdes$IntegerSerde

Properties default.list.key/value.list.* will be ignored as long as default.key/value.serde is not set to org.apache.kafka.common.serialization.Serdes$ListSerde

...