Thanks for the KIP Boyang, I made a pass over the KIP and the PR and have some comments:
1. About the public API, I agree with Matthias that we can consider exposing the `innerDeserializer` and `innerSerializer` in the Time/SessionWindowSerializer/Deserializer, and the `innerSerde` in ` WrapperSerde` so that users can still pass in a `WindowedSerde<T>` into Materialized and Consumed. So that we can have the public API as: ``` (final String topic, final Consumed<Windowed<K>, V> consumed, final Materialized<Windowed<K>, V, WindowStore<Bytes, byte[]>> materialized); (final String topic, final Consumed<Windowed<K>, V> consumed); (final String topic, final Materialized<Windowed<K>, V, WindowStore<Bytes, byte[]>> materialized); ``` 2. There is another WIP interface change to introduce a WindowedKTable<K, V> as an alias of KTable<Windowed<K>, V> for a different purpose of adding some functions only allowed for windowed table. I'm wondering with this interface class if we can work around the Java "method has same erasure" error with the same function name? This is just a wild thought, and I think if we ended up adding `Windowed` into the parameters it may not matter about the signature anyways. 3. This is just a question about your use case: it seems in your scenarios, you will materialize the window store twice in your topology: first time when you generate the windowed KTable from an windowed aggregation operator, the aggregation result i.e. the KTable<Windowed<K>, V> is already materialized into a store, and then when you pipe the changelog of this windowed KTable through an intermediate topic, and read from this topic to form a KTable<Windowed<K>, V>, you will materialize this store again, and the two materialized state stores will contain completely the same data. Have you thought about whether you really need to materialize it twice? Guozhang On Sun, Jun 10, 2018 at 3:04 PM, Matthias J. Sax <matth...@confluent.io> wrote: > Thanks lot for the KIP. The general idea to allow reading > windowed-KTables is very useful! Couple of initial comments/question: > > > > About only adding a single `windowedTable()` with no overloads: > > - retention time is no mandatory parameter and we can always use the > default of 1 day > > - instead of enforcing both, Consumed and Materialized, we could also > extend WindowedSerde to exposed its wrapped key-Serde; thus, if a user > passes in `WindowedSerde` the inner key-serde can be extracted and users > do not need to pass in the key-Serde explicitly. > > - while I agree that we need to pass in the window-size parameter, I am > not sure if using Materialized is the best way to do this; it seems that > window-size is the only mandatory parameter, thus we might be able to > pass it directly and thus allow to make `Consumed` and `Materialized` > optional. Something like: > > > windowedTable(String topicName, long windowSizeMs); > > As an (maybe better) alternative, we could also introduce a public > `Windowed` interface similar to `Produced`, `Consumed`, `Materialized` > etc that we us to pass in window parameters. Or maybe reuse the existing > `Windows` class (ie, the same definition that is used in > KGroupStream#windowedBy() can be passed into the new `windowedTable()` > method. > > > > I also noticed, that the KIP only covers TimesWindows. Should we extend > it to cover SessionWindows, too? > > > > > One side effect is that we bring `ChangeLoggingWindowBytesStore` > public for unit test purpose. > > No need to mention this, because this class in in package "internals" > and not part of public API. > > > > > -Matthias > > > > On 5/24/18 10:39 PM, Boyang Chen wrote: > > Hey friends, > > > > > > I know this is critical time for the 2.0 release. Just want to call out > again for further review on the API format. Any feedback would be > appreciated, thank you! > > > > > > Boyang > > > > ________________________________ > > From: Liquan Pei <liquan...@gmail.com> > > Sent: Tuesday, May 22, 2018 4:29 AM > > To: dev@kafka.apache.org > > Subject: Re: [DISCUSSION] KIP-300: Add Windowed KTable API in > StreamsBuilder > > > > This KIP makes sharing a WindowedKTable among Kafka Stream jobs very > easy. > > It would be nice to get this into trunk soon. > > > > Best, > > Liquan > > > > On Mon, May 21, 2018 at 12:25 PM, Boyang Chen <bche...@outlook.com> > wrote: > > > >> Hey all, > >> > >> > >> I would like to start a discussion thread on KIP 300, which introduces a > >> new API called windowedTable() in StreamsBuilder: > >> > >> https://cwiki.apache.org/confluence/display/KAFKA/KIP- > >> 300%3A+Add+Windowed+KTable+API+in+StreamsBuilder > >> > >> > >> The pull request I'm working on is here: https://github.com/apache/ > >> kafka/pull/5044 > >> > >> > >> I understood that the community is busy working on 2.0 release, but this > >> KIP is really important for our internal use case. So if any of you got > >> time, please focus on clarifying the use case and reaching the > agreement of > >> API. Really appreciate your time! > >> > >> > >> Best, > >> > >> Boyang > >> > >> > >> > > > > > > -- > > Liquan Pei > > Software Engineer, Confluent Inc > > > > -- -- Guozhang
