vcrfxia commented on code in PR #13522:
URL: https://github.com/apache/kafka/pull/13522#discussion_r1171919367
##########
streams/src/test/java/org/apache/kafka/streams/integration/KTableKTableForeignKeyJoinIntegrationTest.java:
##########
@@ -249,62 +272,62 @@ public void doJoinFromLeftThenDeleteLeftEntity() {
@Test
public void doJoinFromRightThenDeleteRightEntity() {
- final Topology topology = getTopology(streamsConfig, materialized ?
"store" : null, leftJoin, rejoin);
+ final Topology topology = getTopology(streamsConfig, materialized ?
"store" : null, leftJoin, rejoin, leftVersioned, rightVersioned);
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 LHS records. This test is all about what
happens when we add/remove RHS records
- left.pipeInput("lhs1", "lhsValue1|rhs1");
- left.pipeInput("lhs2", "lhsValue2|rhs2");
- left.pipeInput("lhs3", "lhsValue3|rhs1");
+ left.pipeInput("lhs1", "lhsValue1|rhs1", baseTimestamp);
+ left.pipeInput("lhs2", "lhsValue2|rhs2", baseTimestamp + 1);
+ left.pipeInput("lhs3", "lhsValue3|rhs1", baseTimestamp + 2);
assertThat(
outputTopic.readKeyValuesToMap(),
is(leftJoin
- ? mkMap(mkEntry("lhs1", "(lhsValue1|rhs1,null)"),
- mkEntry("lhs2", "(lhsValue2|rhs2,null)"),
- mkEntry("lhs3", "(lhsValue3|rhs1,null)"))
- : emptyMap()
+ ? mkMap(mkEntry("lhs1", "(lhsValue1|rhs1,null)"),
+ mkEntry("lhs2", "(lhsValue2|rhs2,null)"),
+ mkEntry("lhs3", "(lhsValue3|rhs1,null)"))
+ : emptyMap()
Review Comment:
Ack, sorry about that. I tried looking to see if I have an auto-reformat
setting on my IDE but everything looks to be disabled. I'll be on the lookout
for this in the future; hopefully it won't be an issue again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
