[jira] [Commented] (KAFKA-18692) Consider to unify KStreamImpl "repartitionRequired" with GraphNode "keyChangingOperation"

["appchemist (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/KAFKA-18692?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17923100#comment-17923100 ] 

appchemist commented on KAFKA-18692:
------------------------------------

I don't have a full grasp of the Streams code yet.
but if you don't mind, can I try that POC with your feedback?

> Consider to unify KStreamImpl "repartitionRequired" with GraphNode "keyChangingOperation"
> -----------------------------------------------------------------------------------------
>
>                 Key: KAFKA-18692
>                 URL: https://issues.apache.org/jira/browse/KAFKA-18692
>             Project: Kafka
>          Issue Type: Improvement
>          Components: streams
>            Reporter: Matthias J. Sax
>            Priority: Minor
>
> In `KStreamImpl`, we use a flag `repartitionRequired` which allows us to track if an operator (or one of its transitive ancestors) did modify the key, so if a key-dependent operation like aggregation or join is executed, we know if we need to insert an auto-repartition step.
> In parallel, we also track key-changing operations in our internal "topology graph" representation (which is use by our topology optimization layer), via GraphNode#keyChangingOperation().
> Thus, we have two independent code path which do a similar thing (note, both are semantically not exactly the same thing, so we need to be careful to get this right; more details below). To avoid subtle bugs, it might be worth to refactor the code, and to unify both.
> The high level idea would be (without me looking into details) to remove `KStreamImpl#repartitionRequired` flag all together, as we already pass in a `GraphNode` into `KStreamImpl` and thus can access `isKeyChangingOperation()`. However, `isKeyChangingOperation()` only tracks if the _current_ node is key-changing, and if it returns `false` we don't know anything about it's ancestors. Thus, semantics are different to `repartitionRequired` flag, which already considers ancestor information. Hence, we might need to traverse the `GraphNode` structure backwards, to verify if a parent did change the key or not (cf `InternalStreamsBuilder#getKeyChangingParentNode()` that I believe we could re-use).
> Another thing to consider is, that some operators like `repartition()` explicit reset `repartitionRequired=false` in a hard-coded way. However, `keyChangingOperation=false` does not carry this information – we don't know if the current operator just does not touch the key, or if the current operator _ensures_ that we are partitioned by the current key. Ie, we need a new way to track this information on `GraphNode` to have a way to stop/break the backward traversal if we hit such a node which does this "reset".
> Overall, I believe it would be good to do this rewrite (based on an educated guess...), but it might also turn out (after some POC coding) that it's not a good idea and it's not worth to do as could add more problems as it solves, and we might also just need to close this ticket as "won't fix". Only a PR could tell for sure.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)