[ https://issues.apache.org/jira/browse/KAFKA-8377?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16852686#comment-16852686 ]
Matthias J. Sax commented on KAFKA-8377: ---------------------------------------- Not sure if I can follow. The decision to materialize or not, is made before we start the application, but while building the `Topology` (please correct my if I am wrong). However, `KTableValueGetterSupplier` is only used at runtime and cannot change the decision we made before any longer? Or do you refer to `KTableImpl#valueGetterSupplier()` – maybe we could do the check there and force materialization of the parent KTable for this case, similar to forcing materialization of source KTables? Btw: while digging into the code, I found that we have classes `KTableMaterializedValueGetterSupplier` and `KTableSourceValueGetterSupplier` that are basically the same – might be worth to remove one as side cleanup with this ticket. > KTable#transformValue might lead to incorrect result in joins > ------------------------------------------------------------- > > Key: KAFKA-8377 > URL: https://issues.apache.org/jira/browse/KAFKA-8377 > Project: Kafka > Issue Type: Bug > Components: streams > Affects Versions: 2.0.0 > Reporter: Matthias J. Sax > Priority: Major > Labels: newbie++ > > Kafka Streams uses an optimization to not materialize every result KTable. If > a non-materialized KTable is input to a join, the lookup into the table > results in a lookup of the parents table plus a call to the operator. For > example, > {code:java} > KTable nonMaterialized = materializedTable.filter(...); > KTable table2 = ... > table2.join(nonMaterialized,...){code} > If there is a table2 input record, the lookup to the other side is performed > as a lookup into materializedTable plus applying the filter(). > For stateless operation like filter, this is safe. However, > #transformValues() might have an attached state store. Hence, when an input > record r is processed by #transformValues() with current state S, it might > produce an output record r' (that is not materialized). When the join later > does a lookup to get r from the parent table, there is no guarantee that > #transformValues() again produces r' because its state might not be the same > any longer. > Hence, it seems to be required, to always materialize the result of a > KTable#transformValues() operation if there is state. Note, that if there > would be a consecutive filter() after tranformValue(), it would also be ok to > materialize the filter() result. Furthermore, if there is no downstream > join(), materialization is also not required. > Basically, it seems to be unsafe to apply `KTableValueGetter` on a stateful > #transformValues()` operator. -- This message was sent by Atlassian JIRA (v7.6.3#76005)