[
https://issues.apache.org/jira/browse/KAFKA-9244?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16984048#comment-16984048
]
Kin Siu edited comment on KAFKA-9244 at 11/28/19 2:37 AM:
----------------------------------------------------------
mkEntry("lhs1", "(lhsValue1|rhs2,rhsValue2)") <-- this is a line to check the
State store of join result, since my change of "rhs1" shouldn't generate value,
we should have same state store value as before.
And above test case failed on the point of last "is(emptyMap())" check, failure
message of JUnit test as below :
{code:java}
java.lang.AssertionError:
Expected: is <{}>
but: was <{lhs1=(lhsValue1|rhs2,rhsValue1Delta)}>
Expected :is <{}>
Actual :<{lhs1=(lhsValue1|rhs2,rhsValue1Delta)}>
<Click to see difference>
at org.hamcrest.MatcherAssert.assertThat(MatcherAssert.java:18)
at org.hamcrest.MatcherAssert.assertThat(MatcherAssert.java:6)
at
org.apache.kafka.streams.integration.KTableKTableForeignKeyJoinIntegrationTest.doJoinFromLeftThenChangeForeignKeyMappingOfLeft(KTableKTableForeignKeyJoinIntegrationTest.java:142)
{code}
was (Author: nikuis):
mkEntry("lhs1", "(lhsValue1|rhs2,rhsValue2)") <-- this is a line to check the
State store of join result, since my change of "rhs1" shouldn't generate value,
we should have same state store value as before.
And above test case failed on the point of last "is(emptyMap())" check, below
the failure message of JUnit test
{code:java}
java.lang.AssertionError:
Expected: is <{}>
but: was <{lhs1=(lhsValue1|rhs2,rhsValue1Delta)}>
Expected :is <{}>
Actual :<{lhs1=(lhsValue1|rhs2,rhsValue1Delta)}>
<Click to see difference>
at org.hamcrest.MatcherAssert.assertThat(MatcherAssert.java:18)
at org.hamcrest.MatcherAssert.assertThat(MatcherAssert.java:6)
at
org.apache.kafka.streams.integration.KTableKTableForeignKeyJoinIntegrationTest.doJoinFromLeftThenChangeForeignKeyMappingOfLeft(KTableKTableForeignKeyJoinIntegrationTest.java:142)
{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)
