[
https://issues.apache.org/jira/browse/KAFKA-21357?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Siddhartha Devineni reassigned KAFKA-21357:
-------------------------------------------
Assignee: Siddhartha Devineni
> Topology optimization fails with "Found a null keyChangingChild node" on
> nested merges
> --------------------------------------------------------------------------------------
>
> Key: KAFKA-21357
> URL: https://issues.apache.org/jira/browse/KAFKA-21357
> Project: Kafka
> Issue Type: Bug
> Components: streams
> Affects Versions: 4.3.0, 4.4.0
> Reporter: Nicholas Telford
> Assignee: Siddhartha Devineni
> Priority: Minor
>
> h3. Problem
> When {{topology.optimization=all}} is set, {{StreamsBuilder.build()}} throws
> this exception {*}for a topology with nested merges{*}:
> {noformat}
> org.apache.kafka.streams.errors.StreamsException: Found a null
> keyChangingChild node for OptimizableRepartitionNode{ BaseRepartitionNode{...
> repartitionTopic='grp-repartition', ...}
> StreamsGraphNode{nodeName='grp-repartition-source', ..., parentNodes=[]} }
> {noformat}
> The failure occurs when all of these conditions are true:
> # A key-changing operation feeds a merge (the inner merge).
> # The inner merge feeds a second merge (the outer merge).
> # An optimizable repartition (for example {{{}groupByKey().count(){}}})
> follows the outer merge.
> # No other key-changing operation sits between them.
> The same topology builds without an error on 4.2.2 and earlier.
> h3. Reproduction
> {code:java}
> StreamsBuilder b = new StreamsBuilder();
> KStream<String, String> f1 = b.<String, String>stream("a").selectKey((k, v)
> -> v);
> KStream<String, String> f2 = b.<String, String>stream("b").selectKey((k, v)
> -> v);
> KStream<String, String> j = b.<String, String>stream("c").selectKey((k, v)
> -> v);
> f1.merge(f2).merge(j)
> .groupByKey(Grouped.with("grp", Serdes.String(), Serdes.String()))
> .count();
> Properties p = new Properties();
> p.put(StreamsConfig.APPLICATION_ID_CONFIG, "repro");
> p.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
> p.put(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG, StreamsConfig.OPTIMIZE);
> b.build(p); // throws on 4.3.x and 4.4.0
> {code}
> ||Version||Result||
> |4.1.2|OK|
> |4.2.2|OK|
> |4.3.1|StreamsException|
> |4.4.0|StreamsException|
> A single merge with three inputs (a merge node with three parents) builds
> without an error on 4.4.0.
> h3. Cause
> KAFKA-20264 changed {{{}InternalStreamsBuilder.findParentNodeMatching{}}}.
> The method now returns the first match in a depth-first search. Before the
> change, it returned the result of the last parent. The change is correct, but
> it exposes a defect in {{{}maybeUpdateKeyChangingRepartitionNodeMap{}}}.
> In the reproduction, M1 = {{{}f1.merge(f2){}}}, M2 = {{{}M1.merge(j){}}}, and
> R is the repartition node below M2.
> # {{getKeyChangingParentNode(R)}} searches up through M2, then M1, and
> returns f1. Before KAFKA-20264, it returned j, the last parent of M2. R is
> now registered under f1.
> # {{maybeUpdateKeyChangingRepartitionNodeMap}} processes each merge node
> separately. R is downstream of both M1 and M2, and f1 is upstream of both. So
> both M1 and M2 take a copy of f1's set, and R goes into two sets.
> # {{mergeRepartitionTopics}} processes the M1 set first. It replaces R and
> removes R from its parents, so {{R.parentNodes()}} becomes empty.
> # It then processes the M2 set, which also contains R.
> {{findParentNodeMatching(R, ...)}} finds no parents and returns {{{}null{}}},
> and the exception occurs.
> Before KAFKA-20264, R was registered under j. M1 is not downstream of j, so
> only M2 took the set, and R was processed one time.
> h3. Suggested fix
> Make sure that each {{OptimizableRepartitionNode}} goes into only one merge
> node's set. Two possible ways:
> * In {{{}maybeUpdateKeyChangingRepartitionNodeMap{}}}, give each key
> changer's set only to the merge node nearest to the repartition nodes.
> * In {{{}mergeRepartitionTopics{}}}, skip a repartition node that has
> already been processed.
> The tests that KAFKA-20264 added cover only a single merge. A test for nested
> merges would catch this regression.
> h3. Workaround
> Do not nest merges above a repartition. Merge all inputs at one merge node.
> The public API has only a two-input {{{}merge{}}}, so this requires a custom
> N-way merge.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
