Skip to content

KAFKA-8705: Remove parent node after leaving loop to prevent NPE #7117

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
bbejeck merged 1 commit into from
Dec 12, 2019
Merged

KAFKA-8705: Remove parent node after leaving loop to prevent NPE #7117

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 7 additions & 2 deletions streams/src/main/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
Expand Down Expand Up @@ -378,6 +379,7 @@ private void maybeOptimizeRepartitionOperations() {

private void maybeUpdateKeyChangingRepartitionNodeMap() {
final Map<StreamsGraphNode, Set<StreamsGraphNode>> mergeNodesToKeyChangers = new HashMap<>();
final Set<StreamsGraphNode> mergeNodeKeyChangingParentsToRemove = new HashSet<>();
for (final StreamsGraphNode mergeNode : mergeNodes) {
mergeNodesToKeyChangers.put(mergeNode, new LinkedHashSet<>());
final Collection<StreamsGraphNode> keys = keyChangingOperationsToOptimizableRepartitionNodes.keySet();

@abbccdda abbccdda Jul 29, 2019

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The maps' names involved are very lengthy such as keyChangingOperationsToOptimizableRepartitionNodes, mergeNodeKeyChangingParentsToRemove, , shall we comment on their individual functionalities?

@bbejeck bbejeck Jul 31, 2019

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ack

Expand All @@ -395,11 +397,14 @@ private void maybeUpdateKeyChangingRepartitionNodeMap() {
final LinkedHashSet<OptimizableRepartitionNode> repartitionNodes = new LinkedHashSet<>();
for (final StreamsGraphNode keyChangingParent : keyChangingParents) {
repartitionNodes.addAll(keyChangingOperationsToOptimizableRepartitionNodes.get(keyChangingParent));

@mjsax mjsax Jul 30, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

@bbejeck bbejeck Jul 31, 2019

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

keyChangingOperationsToOptimizableRepartitionNodes.remove(keyChangingParent);
mergeNodeKeyChangingParentsToRemove.add(keyChangingParent);
}

keyChangingOperationsToOptimizableRepartitionNodes.put(mergeKey, repartitionNodes);
}

for (final StreamsGraphNode mergeNodeKeyChangingParent : mergeNodeKeyChangingParentsToRemove) {

@abbccdda abbccdda Jul 29, 2019

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same here, add a comment on why we need to remove these nodes

@bbejeck bbejeck Jul 31, 2019

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ack

keyChangingOperationsToOptimizableRepartitionNodes.remove(mergeNodeKeyChangingParent);
}
}

@SuppressWarnings("unchecked")
Expand Down
49 changes: 49 additions & 0 deletions streams/src/test/java/org/apache/kafka/streams/kstream/internals/graph/StreamsGraphTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.kstream.Consumed;
import org.apache.kafka.streams.kstream.Grouped;
import org.apache.kafka.streams.kstream.JoinWindows;
import org.apache.kafka.streams.kstream.KStream;
Expand Down Expand Up @@ -124,6 +125,28 @@ public void shouldNotOptimizeWhenAThroughOperationIsDone() {

}

@Test
public void shouldOptimizeSeveralMergeNodesWithCommonKeyChangingParent() {
final StreamsBuilder streamsBuilder = new StreamsBuilder();
final KStream<Integer, Integer> parentStream = streamsBuilder.stream("input_topic", Consumed.with(Serdes.Integer(), Serdes.Integer()))
.selectKey(Integer::sum);

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);

@mjsax mjsax Oct 22, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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).


childStream1
.merge(childStream2)
.merge(childStream3)
.to("output_topic");

@mjsax mjsax Jul 30, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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?

@bbejeck bbejeck Jul 31, 2019

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

@mjsax mjsax Oct 22, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we can remove the to() operator to verify if we don't fail with a NPE.


final Properties properties = new Properties();
properties.setProperty(StreamsConfig.TOPOLOGY_OPTIMIZATION, StreamsConfig.OPTIMIZE);
final Topology topology = streamsBuilder.build(properties);

assertEquals(expectedMergeOptimizedTopology, topology.describe().toString());
}

private Topology getTopologyWithChangingValuesAfterChangingKey(final String optimizeConfig) {

final StreamsBuilder builder = new StreamsBuilder();
Expand Down Expand Up @@ -242,4 +265,30 @@ private int getCountOfRepartitionTopicsFound(final String topologyString) {
+ " Sink: KSTREAM-SINK-0000000009 (topic: output-topic)\n"
+ " <-- KSTREAM-MAPVALUES-0000000008\n\n";


private String expectedMergeOptimizedTopology = "Topologies:\n" +
" Sub-topology: 0\n" +
" Source: KSTREAM-SOURCE-0000000000 (topics: [input_topic])\n" +
" --> KSTREAM-KEY-SELECT-0000000001\n" +
" Processor: KSTREAM-KEY-SELECT-0000000001 (stores: [])\n" +
" --> KSTREAM-MAPVALUES-0000000002, KSTREAM-MAPVALUES-0000000003, KSTREAM-MAPVALUES-0000000004\n" +
" <-- KSTREAM-SOURCE-0000000000\n" +
" Processor: KSTREAM-MAPVALUES-0000000002 (stores: [])\n" +
" --> KSTREAM-MERGE-0000000005\n" +
" <-- KSTREAM-KEY-SELECT-0000000001\n" +
" Processor: KSTREAM-MAPVALUES-0000000003 (stores: [])\n" +
" --> KSTREAM-MERGE-0000000005\n" +
" <-- KSTREAM-KEY-SELECT-0000000001\n" +
" Processor: KSTREAM-MAPVALUES-0000000004 (stores: [])\n" +
" --> KSTREAM-MERGE-0000000006\n" +
" <-- KSTREAM-KEY-SELECT-0000000001\n" +
" Processor: KSTREAM-MERGE-0000000005 (stores: [])\n" +
" --> KSTREAM-MERGE-0000000006\n" +
" <-- KSTREAM-MAPVALUES-0000000002, KSTREAM-MAPVALUES-0000000003\n" +
" Processor: KSTREAM-MERGE-0000000006 (stores: [])\n" +
" --> KSTREAM-SINK-0000000007\n" +
" <-- KSTREAM-MERGE-0000000005, KSTREAM-MAPVALUES-0000000004\n" +
" Sink: KSTREAM-SINK-0000000007 (topic: output_topic)\n" +
" <-- KSTREAM-MERGE-0000000006\n\n";

}