[KAFKA-9321] StreamsBuilder table method overwrites the materialized parameter

• Type: Improvement
• Status: Resolved
• Priority: Major
• Resolution: Duplicate
• Affects Version/s: 2.3.0, 2.4.0
• Component/s: streams
• Assignee: Unassigned
• Reporter: Stefan Bocutiu
• Created: 19/Dec/19 16:34
• Updated: 19/Dec/19 17:42
• Resolved: 19/Dec/19 17:39

Description

One of the methods on the StreamsBuilder to create a KTable takes 3 arguments:

// code placeholder

public synchronized <K, V> KTable<K, V> table(final String topic,
                                              final Consumed<K, V> consumed,
                                              final Materialized<K, V, KeyValueStore<Bytes, byte[]>> materialized)

The method code, however, overwrites the materialized Key & Value serde with the ones coming from consumed. There is already  .table(String, Consumed)  which constructs the table this way. 

 

// code placeholder
...
final ConsumedInternal<K, V> consumedInternal = new ConsumedInternal<>(consumed);
materialized.withKeySerde(consumedInternal.keySerde()).withValueSerde(consumedInternal.valueSerde());

This stops use cases where a table should consume let's say Json but materialize it as Avro. 

 

Suggestion

Drop the line overwriting the Key&Value serdes and ideally remove the `withKeySerde` and `withValueSerde` from Materialized - it could help avoid scenarios like this where state is changed unexpectedly and incorrectly.

 

Issue Links

• duplicates: KAFKA-7777 Decouple topic serdes from materialized serdes