KAFKA-8705: Remove parent node after leaving loop to prevent NPE - #7117
Conversation
|
ping @guozhangwang, @mjsax, @vvcephei, @ableegoldman, @cadonna, @abbccdda |
abbccdda
left a comment
abbccdda
left a comment
There was a problem hiding this comment.
Thanks for the PR. Have we verified that the NPE will recur without this fix in unit test?
| final Set<StreamsGraphNode> mergeNodeKeyChangingParentsToRemove = new HashSet<>(); | ||
| for (final StreamsGraphNode mergeNode : mergeNodes) { | ||
| mergeNodesToKeyChangers.put(mergeNode, new LinkedHashSet<>()); | ||
| final Collection<StreamsGraphNode> keys = keyChangingOperationsToOptimizableRepartitionNodes.keySet(); |
There was a problem hiding this comment.
The maps' names involved are very lengthy such as keyChangingOperationsToOptimizableRepartitionNodes, mergeNodeKeyChangingParentsToRemove, , shall we comment on their individual functionalities?
| keyChangingOperationsToOptimizableRepartitionNodes.put(mergeKey, repartitionNodes); | ||
| } | ||
|
|
||
| for (final StreamsGraphNode mergeNodeKeyChangingParent : mergeNodeKeyChangingParentsToRemove) { |
There was a problem hiding this comment.
Same here, add a comment on why we need to remove these nodes
|
Marking that we have already got 3/3 green check. Future flaky tests shall be ignored for comment-only changes. |
| childStream1 | ||
| .merge(childStream2) | ||
| .merge(childStream3) | ||
| .to("output_topic"); |
There was a problem hiding this comment.
Seems I don't understand the root cause of the bug. Don't we need to add a "key dependent" operation (like aggregation) to trigger a repartition and to merge the nodes to only create one repartition topic?
There was a problem hiding this comment.
Going back to my comment above, when users specify OPTIMIZE the code eagerly makes changes for adjusting the graph for any merge nodes that may require repartitioning. But in this case, we should do nothing as there is nothing to trigger a repartition. I'll look to fix this, but I don't see this being a quick fix, so my preference is to do this in a follow-up PR.
There was a problem hiding this comment.
I think we can remove the to() operator to verify if we don't fail with a NPE.
| @@ -395,11 +397,14 @@ private void maybeUpdateKeyChangingRepartitionNodeMap() { | |||
| final LinkedHashSet<OptimizableRepartitionNode> repartitionNodes = new LinkedHashSet<>(); | |||
| for (final StreamsGraphNode keyChangingParent : keyChangingParents) { | |||
| repartitionNodes.addAll(keyChangingOperationsToOptimizableRepartitionNodes.get(keyChangingParent)); | |||
There was a problem hiding this comment.
The stack trace on the ticket indicate that the NPE is thrown in this line, because keyChangingOperationsToOptimizableRepartitionNodes.get(keyChangingParent) returns null -- can you elaborate why it becomes null?
From my current understanding, keyChangingParents should be a Set and hence we should get each item once and remove it once from keyChangingOperationsToOptimizableRepartitionNodes (in the old code). Why would we get() the same item twice? Seems I am missing something.
There was a problem hiding this comment.
This is due to a single key changing node having multiple merge child nodes. We build a map where the keys are merge nodes and point to a collection of key changing parents, hence we need to wait to remove from the map until we've dropped out of the loop. cf https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilder.java#L380-L389
I'm not completely sure of this design now, I'm revisiting it at the moment, but IMHO it's beyond the scope of this PR.
Yes the test fails without the fix. |
|
ping @mjsax |
|
|
||
| final KStream<Integer, Integer> childStream1 = parentStream.mapValues(v -> v + 1); | ||
| final KStream<Integer, Integer> childStream2 = parentStream.mapValues(v -> v + 2); | ||
| final KStream<Integer, Integer> childStream3 = parentStream.mapValues(v -> v + 3); |
There was a problem hiding this comment.
While this is the example from the ticket, the issue is not related to branching out as done here. Should we rewrite/simplify the test and just create 3 KStream sX = streams.builder() and later add one single key-changing operation? -> s1.selectKey().merge(s2).merge(s3).
|
@abbccdda Call for second review. |
|
Merged #7117 into trunk |
|
will cherry-pick to 2.4 once next release is out |
Fixes case where multiple children merged from a key-changing node causes an NPE. Reviewers: Matthias J. Sax <mjsax@apache.org>, Boyang Chen <boyang@confluent.io>
Fixes case where multiple children merged from a key-changing node causes an NPE. Reviewers: Matthias J. Sax <mjsax@apache.org>, Boyang Chen <boyang@confluent.io>
|
cherry-picked to 2.3 and 2.2 |
Fixes case where multiple children merged from a key-changing node causes an NPE. Reviewers: Matthias J. Sax <mjsax@apache.org>, Boyang Chen <boyang@confluent.io>
|
cherry-picked to 2.4 |
…che#7117) Fixes case where multiple children merged from a key-changing node causes an NPE. Reviewers: Matthias J. Sax <mjsax@apache.org>, Boyang Chen <boyang@confluent.io>
3 participants
Fixes case where multiple children merged from a key-changing node causes an NPE.
I've updated the tests.
Committer Checklist (excluded from commit message)