...

Public Interfaces

• 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 on ListSerializer and ListDeserializer classes
• New method public static <T> Serde<List<T>> <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>> <T> Serde<List<T>> ListSerde(Class listClass, Serde<T> Serde<T> innerSerde) in org.apache.kafka.common.serialization.Serdes class

Proposed Changes

This KIP proposes adding new ListSerializer and ListDeserializer classes as well as support for the new classes into the Serdes class. This will allow using List<T> Serde directly from Consumers, Producers and Streams.

List<T> serialization and deserialization will be done through repeatedly calling a serializer/deserializer for each entry provided by passed generic T's Serde. For example, if you want to create List of Strings serde, then serializer/deserializer of Serdes.StringSerde will be used to serialize/deserialize each entry in `List<String>` List<String>.

Proposed Configurations

List serde is an unusual type of serde because we need to consider two things here: the implementation of List  List (i.e. ArrayList, LinkedList, etc) and enclosed elements' type.

...

1. If the inner serde has one of the following serializers (LongSerializer.class, IntegerSerializer.class, ShortSerializer.class, FloatSerializer.class, DoubleSerializer.class), then the final payload will not contain each element's size since the size of listed types is static (8 bytes, 4 bytes, 2 bytes, etc)
2. If the inner serde doesn't have one of the serializers listed above, then a size of each element will be encoded in the final payload (see below)

...