[KAFKA-9260] Improve Serde "push down" and "wrapping"

• Type: Improvement
• Status: Open
• Priority: Major
• Resolution: Unresolved
• Component/s: streams
• Assignee: Unassigned
• Reporter: Matthias J. Sax
• Created: 02/Dec/19 23:37
• Updated: 09/Dec/19 03:12

Description

Kafka Streams DSL supports "serde push down" feature to let downstream operators inherit upstream serdes if key and/or value are not modified. Furthermore, some operators "wrap" user specified serdes internally (eg, windowed aggregations wrap the user specified key-serde with a time/session window serde – some stores use `ValueAndTimestampSerde` and foreign-key joins also uses some internal wrappers).

The current implementation faces couple of issues, because the "serde push down" feature is a DSL level feature that is used when the Topology is generated. Furthermore, "serde wrapping" is an operator specific feature, not a DSL concept per-se. At runtime, neither "push down" nor "wrapping" are know concepts.

This design implies that if users specify serdes, wrapping and push down works as expected. However, if we fall back to default serdes, special care needs to be taken: for example, some operators not apply the wrapping logic during translation time, and there is additional code that does the wrapping of default serdes as runtime. Another approach would be to wrap a null-Serde, and update the wrapper later (ie, overwrite `null` with the default serde from the config).

Overall, the current design leads to bugs (eg, KAFKA-9248 and KAFKA-9259), and user confusion how it actually works and when/where to specify serdes. Hence, we should consider to rework how we do serde push down and/or wrapping.