[ https://issues.apache.org/jira/browse/KAFKA-9244?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Kin Siu updated KAFKA-9244: --------------------------- Description: 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} was: 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} @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} > 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)