Nicholas Telford created KAFKA-21357:
----------------------------------------
Summary: 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
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)
