Table of Contents |
---|
Status
Current state: Under DiscussionAccepted
Discussion thread: Not available now here
JIRA: here
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
...
We have an existing table() API in the StreamsBuilder which could materialize a Kafka topic into a local state store called KTable. This interface is very useful when we want to back up a Kafka topic to local store. As we know, currently there are 2 different types of state store: key-value based and window based. Sometimes we have certain requirement to materialize a windowed topic (or changlog topic) created by another Stream application into local store, too. The current interface could only accept key-value store, which is not ideal. There are certain cases we need to materialize a windowed topic (or we call changlog topic) created by another Stream application into local store. In this KIP, we would like to address this problem by creating a new API called windowedTable() which supports the generation new APIs to support the materialization of a windowed KTable, storing as either window store or session store.
Here comes the The tricky part : is that when building this API, in the source processor point of view, the windowed topic input should be (Windowed<K> key, V value). So there are actually Note that this is different from a normal topic as the serdes required here should be windowed serdes. Let's clear the four different cases involved in this KIPthe discussion:
- Non-windowed topic materialized to key-value store. This is the most common case and has already been covered by table() API API.
- Non-windowed topic materialized to window store. This is a fallacious requirement because we could easily use aggregate() API to generate a window store based on non-windowed topic.
- Windowed topic (stream KStream changelog) materialized to key-value store. This is also a rare requirement to discuss, because the natural difference between key-value store and window store is that window store sets a retention of the data. By materializing windowed topic to key-value we lost the control on the TTL, which would not be the correct leads to wrong representation of the changelog data.
- Windowed topic (stream KStream changelog) materialized to window store. This is a missing requirement which needs to be addressed by our new API. Currently it's very hard to share a changlog changelog between stream applications, and it could be really useful to share the same state store across applications by this API.
Public Interfaces
The current KTable API looks likeAPIs are defined in the StreamsBuilder class:
Code Block | ||||
---|---|---|---|---|
| ||||
public synchronized <K, V> KTable<K, V> table(final String topic);, final Consumed<K, V> consumed, final Materialized<K, V, KeyValueStore<Bytes, byte[]>> materialized) public synchronized <K, V> KTable<K, V> table(final String topic, final Consumed<K, V> consumed); public synchronized <K, V> KTable<K, V> table(final String topic, final Materialized<KConsumed<K, V, KeyValueStore<Bytes, byte[]>> materializedV> consumed); public synchronized <K, V> KTable<K, V> table(final String topic, final Consumed<K, V> consumed, final Materialized<K, V, KeyValueStore<Bytes, byte[]>> materialized); |
Through Materialized `Materialized` struct, we could pass in a KeyValueStore<Bytes, byte[]> struct as the local state store. In fact, underlying KTable class by default stores data in a key-value store backed up by RocksDB. We want to also support window store which is a very natural requirement if we are materializing a windowed topic with windowed key.
Proposed Changes
We would like to add 4 8 new APIs to support window store and session store as underlying storage option for windowed topic.
Code Block | ||||
---|---|---|---|---|
| ||||
// New APIs: window store materialization to time windowed KTable public synchronized <K, V> KTable<Windowed<K>, V> windowedTable(final String topic, final Duration windowSize, final Consumed<K, V> consumed, final Materialized<K, V, WindowStore<Bytes, byte[]>> materialized); public synchronized <K, V> KTable<Windowed<K>, V> windowedTable(final String topic, final Duration windowSize); public synchronized <K, V> KTable<Windowed<K>, V> windowedTable(final String topic, final Consumed<Windowed<K>Duration windowSize, final Consumed<K, V> consumed); public synchronized <K, V> KTable<Windowed<K>, V> windowedTable(final String topic, final Duration windowSize, final Materialized<K, V, WindowStore<Bytes, byte[]>> materialized); // New APIs: session store materialization to session windowed KTable public synchronized <K, V> KTable<Windowed<K>, V> sessionedTable(final String topic, final Consumed<K, V> windowedTable consumed, final Materialized<K, V, SessionStore<Bytes, byte[]>> materialized); public synchronized <K, V> KTable<Windowed<K>, V> sessionedTable(final String topic); public synchronized <K, V> KTable<Windowed<K>, V> sessionedTable(final String topic, final Consumed<Windowed<K>Consumed<K, V> consumed); public synchronized <K, V> KTable<Windowed<K>, V> sessionedTable(final String topic, final Materialized<K, V, WindowStore<BytesSessionStore<Bytes, byte[]>> materialized); |
As the new API suggests, we are tailing from a windowed changelog topic to materialize the data as a KTable of type <Windowed<K>, V> for processing. Internally, the Consumed struct will be converted to <Windowed<K>, V> to correctly deserialize the changelog records. For Materialized the serde type will still be <K, V> because the window store needs raw key serdes and automatically wrapped with windowed key serde (Checkout WindowedKeySchema.toStoreKeyBinary). These details however, are hided from end user. After KIP-393 we have built the constructor which could wrap around a general key serde to make it a window serde, so stream user doesn't need to worry about the type casting, providing raw key serdes should be suffice.
A `windowSize` duration is required to properly initialize the time windowed serde. This is because the underlying storage for windowed records are only storing window start timestamp for space efficiency. When using the new time windowed API, it is required to explicitly pass in positive window size for initialization. User must be aware of the windowed topic window size in order to properly deserialize the topic. For session window serde, `windowSize` config is not needed, because we don't know the individual window size beforehand, so each record will store both start and end time.
Compatibility, Deprecation, and Migration Plan
...