[ https://issues.apache.org/jira/browse/KAFKA-13164?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17424296#comment-17424296 ]
Hao Li commented on KAFKA-13164: -------------------------------- Hi Ralph, I was able to test it out with following code snippet: {code:java} static void createWordCountStream(final StreamsBuilder builder) { final KStream<String, String> textLines = builder.stream(joinInputTopic); final KTable<String, String> table = builder.<String, String>table(inputTopic) .mapValues(v -> { System.out.println("Table got " + v); return v + " suffix"; }); final KStream<String, String> output = textLines.leftJoin(table, (sv, tv) -> { System.out.println("Join function called for " + sv + ", " + tv); return sv + tv; }); output.to(outputTopic); } {code} I was able to see the join function got called with correct sv and tv. I also saw table map values function called before join function called. So as [~guozhang] mentioned, the state store is attached to source table and the map value function is called on the fly and then join happens. So that's expected. However, I wasn't able to see output in output topic. That requires more digging but state store attaching to source table seems to be expected. > State store is attached to wrong node in the Kafka Streams topology > ------------------------------------------------------------------- > > Key: KAFKA-13164 > URL: https://issues.apache.org/jira/browse/KAFKA-13164 > Project: Kafka > Issue Type: Bug > Components: streams > Affects Versions: 2.8.0 > Environment: local development (MacOS Big Sur 11.4) > Reporter: Ralph Matthias Debusmann > Assignee: Hao Li > Priority: Major > Fix For: 3.0.1 > > Attachments: 1.jpg, 3.jpg > > > Hi, > mjsax and me noticed a bug where a state store is attached to the wrong node > in the Kafka Streams topology. > The issue arised when I tried to read a topic into a KTable, then continued > with a mapValues(), and then joined this KTable with a KStream, like so: > > var kTable = this.streamsBuilder.table(<topic>).mapValues(<mapValues > function>); > > and then later: > > var joinedKStream = kstream.leftJoin(kTable, <leftJoin function>); > > The join didn't work, and neither did it work when I added Materialized.as() > to mapValues(), like so: > var kTable = this.streamsBuilder.table(<topic>).mapValues(<mapValues > function>, *Materialized.as(<some name>)*); > > Interestingly, I could get the join to work, when I first read the topic > into a *KStream*, then continued with the mapValues(), then turned the > KStream into a KTable, and then joined the KTable with the other KStream, > like so: > > var kTable = this.streamsBuilder.stream(<topic>).mapValues(<mapValues > function>).toTable(); > > (the join worked the same as above) > > When mjsax and me had a look on the topology, we could see that in the > former, not working code, the state store (required for the join) is attached > to the pre-final "KTABLE-SOURCE", and not the final "KTABLE-MAPVALUES" node > (see attachment "1.jpg"). In the working code, the state store is (correctly) > attached to the final "KSTREAM-TOTABLE" node (see attachment "3.jpg"). > > Best regards, > xdgrulez > > > -- This message was sent by Atlassian Jira (v8.3.4#803005)