...
default.key/value.serde = org.apache.kafka.common.serialization.Serdes$ListSerde
Then, we need to cover those introduce two brand new configurations and here I'm proposing these four extra properties:
CommonClientConfigs.class (duplicated in StreamsConfig.class): DEFAULT_LIST_KEY/TYPE_SERDE_TYPE_CLASS = "default.list.key/value.serde.type"
Ex. default = java.util.ArrayList default.list.key/value.serde.inner type = orgjava.apacheutil.kafka.common.serialization.Serdes$IntegerSerdeProperties ArrayList
CommonClientConfigs.class (duplicated in StreamsConfig.class): DEFAULT_LIST_KEY/TYPE_SERDE_INNER_CLASS = "default.list.key.serde.inner"
Ex. default.list./value.* will be ignored as long as default.key/value.serde is not set to org.inner = org.apache.kafka.common.serialization.Serdes$ListSerdeSerdes$IntegerSerde
Other proposed properties:
ConsumerConfig.class: LIST_KEY_DEERIALIZERDESERIALIZER_INNERTYPE_CLASS_CONFIG = "list.key.deserializer.innertype"
ConsumerConfig.class: LIST_KEY_DESERIALIZERDEERIALIZER_TYPEINNER_CLASS_CONFIG = "list.key.deserializer.typeinner"
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 case 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
Serialization Strategy
For the performance purposes the following serialization strategy was put in place. Depending on the type of an inner serde (a list's element type) the serialization will be performed in the following ways:
...