KAFKA-6840: Add windowed-KTable API

[abbccdda]

We are proposing adding a new API called windowedTable to StreamsBuilder.java to bring in the ability of materializing a windowed topic to local. Links to related sources:
KIP: https://cwiki.apache.org/confluence/display/KAFKA/KIP-300%3A+Add+Windowed+KTable+API+in+StreamsBuilder
Jira: https://issues.apache.org/jira/browse/KAFKA-6840

[abbccdda]

Hey @mjsax @guozhangwang, before we start the discussion thread, maybe you could give me some initial feedback on the code or KIP? Especially the logic has been a little deviated from our initial discussion on Jira, and I reflected those in the KIP. Let me know if you need more clarification, thank you!

[mjsax]

Thanks @abbccdda! AK release deadline is coming up and we need to focus on getting pending KIPs and PR ready for the release. We will cycle back when we find time. Thanks for your understanding. (Feel free to ping us after the 2.0 release is out in case this would slip.)

[mjsax]

\cc @bbejeck @vvcephei

[tedyu]

Add Apache license header.

[tedyu]

wrong class name

[abbccdda]

Thanks for the clarification @mjsax so we are expecting 2.0 fully release in June?

[Ishiihara]

nit: format

[Ishiihara]

nit: might be better to name it windowedKTable

[abbccdda]

to be consistent with createKTable()

[mjsax]

+1 to rename to windowedKTable

nit: fit formatting (either move consumed down one line, or indent other parameter to match indention of consumed)

[Ishiihara]

We can define a local variable for key.window().start()

[Ishiihara]

test this please

[tedyu]

Without adding license header, no tests would be run.

[Ishiihara]

@mjsax @guozhangwang Thanks a lot for helping in review the PR. We totally understand the priority of the Kafka 2.0 release and I am wondering if we can do a review on the API and have an agreement? We can leave the detailed review and polish later when you have more time. Really appreciate your help!

[mjsax]

@abbccdda Release plan can be found here: https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=80448820

After code freeze we should have time. Maybe even before. Just can't promise, but I'll try to squeeze it in if possible. :)

[abbccdda]

@tedyu thanks for the comments. Addressing

[abbccdda]

@Ishiihara seems passed.

[abbccdda]

@Ishiihara will resolve conflict tonight

[Ishiihara]

LGTM in general. Left some minor comments.

[Ishiihara]

Do we want to support session window as well?

[abbccdda]

I don't think we would support it for now, maybe in next generation

[Ishiihara]

nit: might be better to set end to another timestamp.

[Ishiihara]

s/Serde/serde

[Ishiihara]

nit: format

[Ishiihara]

Add some comments for the return type.

[Ishiihara]

One question, why this class needs to be public?

[abbccdda]

We need to use it for unit testing purpose

[Ishiihara]

Do we want to test both session window and time window?

[abbccdda]

For internal release, this could be postponed. Will add session window test later

[abbccdda]

@mjsax @guozhangwang we are planning an internal release next week, could you take a quick look on the API to provide us some feedback? Thanks!

[Ishiihara]

@abbccdda One high level comment: it seems that we have quite a few code duplications for the time window and session window. It would be nice if we can reduce the duplication.

[guozhangwang]

This is a meta comment: we do not have unit tests for SessionWindowedKTableSource and SessionWindowedStoreMaterializer, is it intentional?

[guozhangwang]

The other PR #5307 is close to be merged I think. We should proceed on voting and accepting the KIP in order to progress. Then we can come back to this PR after 5307 is merged.

[abbccdda]

Hey guys @guozhangwang @Ishiihara @shawnsnguyen This is a draft version for latest KIP-300. To make sure the review is efficient, be advised to focus on looking at changes in StreamsBuilder, InternalStreamsBuilder, KTableSource and TableSourceNode classes. As you could see this is a very large change, so I want to make sure major changes are getting common understanding before proceeding to further unit test changes.

Major changes include:

1. generalize the #table() API in InternalStreamsBuilder where our return type is <?, V> instead of strict <K, V>. The reasoning is that we want to avoid too much code redundancy and the public API could cast the return type as needed. We couldn't make the type more explicit because underlying window store, session store and key value store are not sharing single common interface.

2. Make KTableSource as abstract class and let different table source (kv, window, session) inherit. This could also save code redundancy.

3. Public API changes to support 8:
   #windowTable(topic, conusmed, materialized)
   #windowTable(topic)
   #windowTable(topic, conusmed)
   #windowTable(topic, materialized)
   #sessionTable(topic, conusmed, materialized)
   #sessionTable(topic)
   #sessionTable(topic, conusmed)
   #sessionTable(topic, materialized)

4. Overload StreamSourceNode to pass in two new parameters:

    private final boolean isWindowed;
    private final ConsumedInternal<Windowed<K>, V> windowedConsumedInteral;

which will replace the returned consumedInternal() to windowed consume. This could reduce the scattering of windowed serde logic and consolidate changes in one central place without changing many class types.

Let me know your thoughts!

[mjsax]

@abbccdda Can you rebase and update this KIP so we can review? Thanks!

[abbccdda]

@mjsax Thanks for taking time to review the change! In fact, I would like to get the vote completed for KIP-300 before proceeding to work on this since the rebasing effort of this change is really high now and we hope this is the last back-and-forth :). Hey @guozhangwang @vahidhashemian could you take a look at KIP-300 and give me two more binding votes when you got time? Thanks!

[mjsax]

@abbccdda I did an initial first pass. It seems this PR is a little outdated compared to the final KIP. Can you update the PR and rebase it to resolve merge conflicts?

\cc @cadonna for review after the PR is updated.

[mjsax]

Why this?

[abbccdda]

Oh, this is my local build failure, will revert lol

[mjsax]

the method name changed to windowedTable and windowSize parameter is missing

[mjsax]

windowSize should be Duration

[abbccdda]

@mjsax Thanks so much for the review Matthias! So sorry I'm currently clearing some backlog tasks at my work and couldn't get chance to react this one. I should be able to start making progress late next week.

[mjsax]

Cool. We are not in a hurry. Just wanted to make sure to unblock you.

[mjsax]

Closing this stale PR -- the corresponding KIP was never approved and the KIP itself seems abandoned atm.