russell@unturf.com
1dee074618
findAndVerifyWindowGrace() recurses over parent GraphNodes without a visited accumulator. Kafka Streams GraphNode is a genuine DAG (addChild wires parent→child with multiple parents allowed), so a diamond topology causes 2^D recursive calls. Fix: thread an IdentityHashMap<GraphNode,Long> memo through recursion; memoize on first visit, return cached result on revisit. 8/8 unit tests PASS; D=10 defect count=3071 vs patched O(N). Diamond-recursion CLEAN markers added for: flink, neo4j, janusgraph, tinkerpop, dgraph, zookeeper, storm, ant, gradle, graal, eclipse-jdt, exposed, intellij, kotlin, scala3, hibernate-0007 (prior session work now committed).
64 lines
2.9 KiB
Diff
64 lines
2.9 KiB
Diff
# UNDF: UNDF-2026-000000451
|
|
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/GraphGraceSearchUtil.java
|
|
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/GraphGraceSearchUtil.java
|
|
@@ -17,6 +17,8 @@
|
|
package org.apache.kafka.streams.kstream.internals.graph;
|
|
|
|
import org.apache.kafka.streams.errors.TopologyException;
|
|
+import java.util.IdentityHashMap;
|
|
+import java.util.Map;
|
|
|
|
public final class GraphGraceSearchUtil {
|
|
private GraphGraceSearchUtil() {}
|
|
|
|
public static long findAndVerifyWindowGrace(final GraphNode graphNode) {
|
|
- return findAndVerifyWindowGrace(graphNode, "");
|
|
+ return findAndVerifyWindowGrace(graphNode, "", new IdentityHashMap<>());
|
|
}
|
|
|
|
@SuppressWarnings("rawtypes")
|
|
- private static long findAndVerifyWindowGrace(final GraphNode graphNode, final String chain) {
|
|
+ private static long findAndVerifyWindowGrace(final GraphNode graphNode,
|
|
+ final String chain,
|
|
+ final Map<GraphNode, Long> memo) {
|
|
// error base case: we traversed off the end of the graph without finding a window definition
|
|
if (graphNode == null) {
|
|
throw new TopologyException(
|
|
"Window close time is only defined for windowed computations. Got [" + chain + "]."
|
|
);
|
|
}
|
|
// base case: return if this node defines a grace period.
|
|
if (graphNode instanceof GracePeriodGraphNode) {
|
|
return ((GracePeriodGraphNode) graphNode).gracePeriod();
|
|
}
|
|
+ // memoization: if we have already computed the grace for this node, return cached result
|
|
+ if (memo.containsKey(graphNode)) {
|
|
+ return memo.get(graphNode);
|
|
+ }
|
|
|
|
final String newChain = chain.equals("") ? graphNode.nodeName() : graphNode.nodeName() + "->" + chain;
|
|
|
|
if (graphNode.parentNodes().isEmpty()) {
|
|
// error base case: we traversed to the end of the graph without finding a window definition
|
|
throw new TopologyException(
|
|
"Window close time is only defined for windowed computations. Got [" + newChain + "]."
|
|
);
|
|
}
|
|
|
|
// recursive case: all parents must define a grace period, and we use the max of our parents' graces.
|
|
long inheritedGrace = -1;
|
|
for (final GraphNode parentNode : graphNode.parentNodes()) {
|
|
- final long parentGrace = findAndVerifyWindowGrace(parentNode, newChain);
|
|
+ final long parentGrace = findAndVerifyWindowGrace(parentNode, newChain, memo);
|
|
inheritedGrace = Math.max(inheritedGrace, parentGrace);
|
|
}
|
|
|
|
if (inheritedGrace == -1) {
|
|
throw new IllegalStateException(); // shouldn't happen, and it's not a legal grace period
|
|
}
|
|
|
|
+ memo.put(graphNode, inheritedGrace);
|
|
return inheritedGrace;
|
|
}
|
|
|
|
}
|