[ https://issues.apache.org/jira/browse/KAFKA-9244?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Matthias J. Sax updated KAFKA-9244: ----------------------------------- Comment: was deleted (was: {quote}mkEntry("lhs1", "(lhsValue1|rhs2,rhsValue2)") <-- this is a line to check the State store of join result, {quote} As I said in my comment above, `outputTopic.readKeyValuesToMap()` does *not* return the _content_ of the state store. Each _update_ to the state store, will write one output record into the output topic, and `outputTopic.readKeyValuesToMap()` will give you those _update_ records. Because the last input record does not trigger an update to the store, the won't be an output record. If you want to check the _content_ of the state store, you would need to access the state store via {quote}{{ToplogyTestDriver#getTimestampedKeyValueStore()}} {{// or if you don't care about the timestamp}} {{ToplogyTestDriver#getKeyValueStore()}}{{}}{quote}) > Update of old FK reference on RHS should not trigger join result > ---------------------------------------------------------------- > > Key: KAFKA-9244 > URL: https://issues.apache.org/jira/browse/KAFKA-9244 > Project: Kafka > Issue Type: Bug > Components: streams > Affects Versions: 2.4.0 > Reporter: Kin Siu > Priority: Major > > Perform a KTable-KTable foreign key join, after changing LHS FK reference > from FK1 -> FK2, populating update on RHS with FK1 should not produce join > result. > Below test case failed at the point of last "is(emptyMap())". > > {code:java|linenumbers=true} > @Test > public void doJoinFromLeftThenChangeForeignKeyMappingOfLeft() { > final Topology topology = getTopology(streamsConfig, "store", > leftJoin); > try (final TopologyTestDriver driver = new > TopologyTestDriver(topology, streamsConfig)) { > final TestInputTopic<String, String> right = > driver.createInputTopic(RIGHT_TABLE, new StringSerializer(), new > StringSerializer()); > final TestInputTopic<String, String> left = > driver.createInputTopic(LEFT_TABLE, new StringSerializer(), new > StringSerializer()); > final TestOutputTopic<String, String> outputTopic = > driver.createOutputTopic(OUTPUT, new StringDeserializer(), new > StringDeserializer()); > final KeyValueStore<String, String> store = > driver.getKeyValueStore("store"); > // Pre-populate the RHS records. This test is all about what > happens when we change LHS records foreign key reference > // then populate update on RHS > right.pipeInput("rhs1", "rhsValue1"); > right.pipeInput("rhs2", "rhsValue2"); > assertThat( > outputTopic.readKeyValuesToMap(), > is(emptyMap()) > ); > assertThat( > asMap(store), > is(emptyMap()) > ); > left.pipeInput("lhs1", "lhsValue1|rhs1"); > { > final Map<String, String> expected = mkMap( > mkEntry("lhs1", "(lhsValue1|rhs1,rhsValue1)") > ); > assertThat( > outputTopic.readKeyValuesToMap(), > is(expected) > ); > assertThat( > asMap(store), > is(expected) > ); > } > // Change LHS foreign key reference > left.pipeInput("lhs1", "lhsValue1|rhs2"); > { > final Map<String, String> expected = mkMap( > mkEntry("lhs1", "(lhsValue1|rhs2,rhsValue2)") > ); > assertThat( > outputTopic.readKeyValuesToMap(), > is(expected) > ); > assertThat( > asMap(store), > is(expected) > ); > } > // Populate RHS update on old LHS foreign key ref > right.pipeInput("rhs1", "rhsValue1Delta"); > { > assertThat( > outputTopic.readKeyValuesToMap(), > is(emptyMap()) > ); > assertThat( > asMap(store), > is(mkMap( > mkEntry("lhs1", "(lhsValue1|rhs2,rhsValue2)") > )) > ); > } > } > } > {code} -- This message was sent by Atlassian Jira (v8.3.4#803005)