[
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)
