Kafka 12317: Relax non-null key requirement in Kafka Streams by florin-akermann · Pull Request #14174 · apache/kafka

florin-akermann · 2023-08-09T12:22:26Z

The key requirments got relaxed for the followinger streams dsl operator:

left join Kstream-Kstream: no longer drop left records with null-key and call ValueJoiner with 'null' for right value.
outer join Kstream-Kstream: no longer drop left/right records with null-key and call ValueJoiner with 'null' for right/left value.
left-foreign-key join Ktable-Ktable: no longer drop left records with null-foreign-key returned by the ForeignKeyExtractor and call ValueJoiner with 'null' for right value.
left join KStream-Ktable: no longer drop left records with null-key and call ValueJoiner with 'null' for right value.
left join KStream-GlobalTable: no longer drop records when KeyValueMapper returns 'null' and call ValueJoiner with 'null' for right value.

Repartitioning:

If a repartition node is upstream to any of the above listed operators then it may not filter null-key-records.
Thus, the new default behavior for repartiton nodes is to no longer filter out null-key-records.
However, we would like to keep this optimization where possible.
Therefore, during the 'buildAndOptimizeTopology' the graph gets traversed to look for repartition nodes which have none of the above listed operators downstream.
These nodes then get updated with the filter perdicate to filter out null-key-records.

Test Strategy:

Tests for each of the targeted dsl operator on minimal topologies plus extending some of the existing Integration tests (e.g. StreamStreamJoinIntegrationTest.java)

Misc:

I suggest updating https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Streams+Join+Semantics after this has been merged.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

florin-akermann · 2023-08-18T17:59:24Z

@mjsax @guozhangwang I'd say this PR is ready to be reviewd. May I ask for your input or do you know who to poke?

wcarlson5 · 2023-10-05T18:48:00Z

Hey @florin-akermann I can take a look at this soon. I'll probably be able to take a first look next week if you are still willing to push this

florin-akermann · 2023-10-05T19:02:12Z

Hey @wcarlson5
Great! Yes I am willing to push it.
Looking forward to your feedback.

wcarlson5

Hey I left some comments, I think thats a good place to start, but I might need to think about a few things more on a second pass. So sorry if I missed something that might come up later.

I didn't have a chance to really dig into the testing yet, but it mostly looked sufficient.

I don't think we are too far off and I hope this lets you make some progress!

wcarlson5 · 2023-10-05T21:12:05Z

This is to prevent null keys to go into reparation topics, right? Will that effect results if a manual reparation is added?

This is related to this change: d1301b1

Previously, as an optimization, null key records were filtered out upon repartitioning.
Now we need null-key records to propagate if there is a left- or outer-join operator further downstream.
@mjsax hinted in KAFKA-12317 that this optimization can no longer be applied everywhere.

So now, by default, the optimization is no longer applied.
rewriteRepartitionNodes adds this optimization if no left- or outer-join operation is downstream to the partitionNode in question.

In general, a manual repartition is covered by this as well.
Though, now I wonder whether there should be some kind of config/flag whether you want this optimization to be applied at all.
Because, you could define some topology with left joins etc. expecting null-key records to propagate all the way through.
However, if you chose to add a manual repartition at the very end of your topology they would then still be filtered out because no left- or outer-join operation is downstream of this last repartition.

I guess developers would like to be able to opt out of this optimization in this case?

Hmm, I would think they would like to opt out. That would require an update to the kip. Maybe even a revote. I'm not sure what the odds are that someone manually repartitioning would be needing the null-keys to propagate. But its probably higher than you would think as manual repartitioner as are typically power-users.

I don't think we need to make it optional as we already filter all null keys and now we let some propagate. Maybe we should just make a ticket and we can come back to it. Being able to toggle the optimization should be pretty simple.

Ok, so basically we remove this optimization completley for now?
Developers could still just filter out null keys with a 'filter' operator to achieve the old behavior.
And then we make a separate ticket where developers can opt in to this optimization?

wcarlson5 · 2023-10-05T21:12:47Z

why should null keys not enter the window?

Should a null key ever match? I thought/think in this context null != null?
Plus, my guess would be that most WindowStore implementations throw upon null as a key?

That's likely true about the windowstore implemntations. Something to think about for later then. We should maybe clarify the semantics about null matching in the docs somewhere.

wcarlson5 · 2023-10-09T19:16:47Z

I'm not a huge fan of splitting this out to a separate public method. I think you can just reuse the logic in skip record.

agree - adjusted

wcarlson5 · 2023-10-09T19:17:32Z

This is the check already in skip record, can you use that instead?

agree - adjusted

florin-akermann

Thank you very much @wcarlson5. I have adjusted the part about the skipRecord and replied to the others.

There is also Jira-14174. I kept it separate as it is a separate Jira.
Would you like me to merge it into this PR?

florin-akermann · 2023-10-10T08:44:43Z

This is related to this change: d1301b1

Previously, as an optimization, null key records were filtered out upon repartitioning.
Now we need null-key records to propagate if there is a left- or outer-join operator further downstream.
@mjsax hinted in KAFKA-12317 that this optimization can no longer be applied everywhere.

So now, by default, the optimization is no longer applied.
rewriteRepartitionNodes adds this optimization if no left- or outer-join operation is downstream to the partitionNode in question.

In general, a manual repartition is covered by this as well.
Though, now I wonder whether there should be some kind of config/flag whether you want this optimization to be applied at all.
Because, you could define some topology with left joins etc. expecting null-key records to propagate all the way through.
However, if you chose to add a manual repartition at the very end of your topology they would then still be filtered out because no left- or outer-join operation is downstream of this last repartition.

I guess developers would like to be able to opt out of this optimization in this case?

florin-akermann · 2023-10-10T09:37:40Z

agree - adjusted

florin-akermann · 2023-10-10T09:37:45Z

agree - adjusted

florin-akermann · 2023-10-10T10:32:50Z

Should a null key ever match? I thought/think in this context null != null?
Plus, my guess would be that most WindowStore implementations throw upon null as a key?

wcarlson5

I think its really close, just a couple of follow ups and we should be done. Thanks for the changes @florin-akermann !

wcarlson5 · 2023-10-19T20:34:21Z

Hmm, I would think they would like to opt out. That would require an update to the kip. Maybe even a revote. I'm not sure what the odds are that someone manually repartitioning would be needing the null-keys to propagate. But its probably higher than you would think as manual repartitioner as are typically power-users.

I don't think we need to make it optional as we already filter all null keys and now we let some propagate. Maybe we should just make a ticket and we can come back to it. Being able to toggle the optimization should be pretty simple.

wcarlson5 · 2023-10-19T20:36:17Z

That's likely true about the windowstore implemntations. Something to think about for later then. We should maybe clarify the semantics about null matching in the docs somewhere.

wcarlson5

Test failures are unrelated

florin-akermann · 2023-11-05T14:06:06Z

@wcarlson5 thanks for the merge. I think it would be good to tackle #14107 in the same release.

On a different note, I noticed that the squashed commit message doesn't adhere to the typically Jira ref: 'Kafka 12317' vs 'KAFKA-12317'.

…14174) [KIP-962](https://cwiki.apache.org/confluence/display/KAFKA/KIP-962%3A+Relax+non-null+key+requirement+in+Kafka+Streams) The key requirments got relaxed for the followinger streams dsl operator: left join Kstream-Kstream: no longer drop left records with null-key and call ValueJoiner with 'null' for right value. outer join Kstream-Kstream: no longer drop left/right records with null-key and call ValueJoiner with 'null' for right/left value. left-foreign-key join Ktable-Ktable: no longer drop left records with null-foreign-key returned by the ForeignKeyExtractor and call ValueJoiner with 'null' for right value. left join KStream-Ktable: no longer drop left records with null-key and call ValueJoiner with 'null' for right value. left join KStream-GlobalTable: no longer drop records when KeyValueMapper returns 'null' and call ValueJoiner with 'null' for right value. Reviewers: Walker Carlson <wcarlson@apache.org>

florin-akermann marked this pull request as draft August 9, 2023 12:22

florin-akermann force-pushed the KAFKA-12317 branch 5 times, most recently from 32c0370 to 88037b5 Compare August 17, 2023 19:48

florin-akermann marked this pull request as ready for review August 17, 2023 21:15

florin-akermann force-pushed the KAFKA-12317 branch 2 times, most recently from edf866a to 0006e20 Compare August 18, 2023 07:41

florin-akermann marked this pull request as draft August 18, 2023 08:07

florin-akermann marked this pull request as ready for review August 18, 2023 17:59

florin-akermann force-pushed the KAFKA-12317 branch 2 times, most recently from 380dbb7 to f17f6b0 Compare September 25, 2023 06:30

wcarlson5 added streams kip Requires or implements a KIP labels Oct 5, 2023

wcarlson5 reviewed Oct 9, 2023

View reviewed changes

florin-akermann commented Oct 10, 2023

View reviewed changes

wcarlson5 reviewed Oct 19, 2023

View reviewed changes

florin-akermann added 9 commits October 25, 2023 21:17

KAFKA-12317: Tests

12ebd65

KAFKA-12317: Don't drop null-key records on repartition

b2f528b

KAFKA-12317: left join KStream-KTable

47bb100

KAFKA-12317: left/outer join KStream-KTable

44f8057

KAFKA-12317: fix code styles

83a5d43

KAFKA-12317: outer join add test case for left key: null

02f0e4d

KAFKA-12317: desired behavior for KStreamGlobalKTable left join

25a8c92

KAFKA-12317: optimize repartition nodes in optimization phase

0933610

KAFKA-12317: upgrade-guide.html

64dfda0

florin-akermann added 8 commits October 25, 2023 21:18

KAFKA-12317: dsl-api.html: detailed behavior

bc0f578

KAFKA-12317: java docs

10f070e

KAFKA-12317: extend KStreamKStream integration test

4c76b18

KAFKA-12317: extend KStreamRepartitionIntegrationTest

17d06ab

KAFKA-12317: extend StreamStreamJoinIntegrationTest

19b50ee

KAFKA-12317: adjust TableTableJoinIntegrationTest

5ab937f

KAFKA-12317: mv node search

601a4a7

KAFKA-12317: reuse skipRecord

ef9128a

florin-akermann force-pushed the KAFKA-12317 branch from 024db29 to ef9128a Compare October 25, 2023 19:30

wcarlson5 approved these changes Oct 31, 2023

View reviewed changes

wcarlson5 merged commit b5c2497 into apache:trunk Oct 31, 2023

florin-akermann mentioned this pull request Dec 3, 2023

Kafka-14748: Relax non-null FK left-join requirement #14107

Merged

3 tasks

Conversation

florin-akermann commented Aug 9, 2023 • edited by lucasbru Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Repartitioning:

Test Strategy:

Misc:

Committer Checklist (excluded from commit message)

Uh oh!

florin-akermann commented Aug 18, 2023

Uh oh!

wcarlson5 commented Oct 5, 2023

Uh oh!

florin-akermann commented Oct 5, 2023

Uh oh!

wcarlson5 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

florin-akermann left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

wcarlson5 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

wcarlson5 left a comment

Choose a reason for hiding this comment

Uh oh!

florin-akermann commented Nov 5, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

florin-akermann commented Aug 9, 2023 •

edited by lucasbru

Loading

florin-akermann left a comment •

edited

Loading

florin-akermann commented Nov 5, 2023 •

edited

Loading