Page History

...

First, we need to specify that we are going to use a list serde:

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

Then, we need to introduce two brand new configurations and here I'm proposing these four extra properties:

CommonClientConfigs.class

...

: DEFAULT_LIST_KEY/

...

VALUE_SERDE_TYPE_CLASS = "default.list.key.serde.type"

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

CommonClientConfigs.class

...

: DEFAULT_LIST_KEY/

...

VALUE_SERDE_INNER_CLASS = "default.list.key.serde.inner"

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

...

Other proposed properties:

ConsumerConfig.class: LIST_KEY_DESERIALIZER_TYPE_CLASS_CONFIG = "list.key.deserializer.type"

...

ProducerConfig.class: LIST_KEY_SERIALIZER_INNER_CLASS_CONFIG = "list.key.serializer.inner"

P.S. We do not need a type class config for the serializer since we do not care about the type of the list class during serialization.P.P.S. Properties default.list.key/value.* will be ignored as long as default.key/value.serde is not set to org.apache.kafka.common.serialization.Serdes$ListSerde

...

1. If an inner serde has one of the following serializers (LongSerializerShortSerializer.class, IntegerSerializer.class, ShortSerializerFloatSerializer.class, FloatSerializer LongSerializer.class, DoubleSerializer.class, UUIDSerializer.class), then the final payload will not contain each element's size encoded since the size sizes of presented types is are static (8 2 bytes, 4  4 bytes, 2 8 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)

...