bellemare commented on a change in pull request #8764:
URL: https://github.com/apache/kafka/pull/8764#discussion_r436871862
##########
File path:
streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableForeignKeyJoinScenarioTest.java
##########
@@ -243,17 +244,17 @@ private void validateTopologyCanProcessData(final
StreamsBuilder builder) {
final String safeTestName = safeUniqueTestName(getClass(), testName);
config.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, "dummy-" +
safeTestName);
config.setProperty(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy");
- config.setProperty(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG,
Serdes.StringSerde.class.getName());
+ config.setProperty(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG,
Serdes.IntegerSerde.class.getName());
config.setProperty(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,
Serdes.StringSerde.class.getName());
config.setProperty(StreamsConfig.STATE_DIR_CONFIG,
TestUtils.tempDirectory().getAbsolutePath());
try (final TopologyTestDriver topologyTestDriver = new
TopologyTestDriver(builder.build(), config)) {
- final TestInputTopic<String, String> aTopic =
topologyTestDriver.createInputTopic("A", new StringSerializer(), new
StringSerializer());
- final TestInputTopic<String, String> bTopic =
topologyTestDriver.createInputTopic("B", new StringSerializer(), new
StringSerializer());
- final TestOutputTopic<String, String> output =
topologyTestDriver.createOutputTopic("output", new StringDeserializer(), new
StringDeserializer());
- aTopic.pipeInput("a1", "b1-alpha");
- bTopic.pipeInput("b1", "beta");
- final Map<String, String> x = output.readKeyValuesToMap();
- assertThat(x, is(Collections.singletonMap("a1",
"(b1-alpha,(b1-alpha,beta))")));
+ final TestInputTopic<Integer, String> aTopic =
topologyTestDriver.createInputTopic("A", new IntegerSerializer(), new
StringSerializer());
+ final TestInputTopic<Integer, String> bTopic =
topologyTestDriver.createInputTopic("B", new IntegerSerializer(), new
StringSerializer());
+ final TestOutputTopic<Integer, String> output =
topologyTestDriver.createOutputTopic("output", new IntegerDeserializer(), new
StringDeserializer());
+ aTopic.pipeInput(1, "1-alpha");
+ bTopic.pipeInput(1, "beta");
Review comment:
Agreed. Good catch.
##########
File path:
streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableForeignKeyJoinScenarioTest.java
##########
@@ -181,60 +182,60 @@ public void shouldWorkWithDefaultAndProducedSerdes() {
public void shouldUseExpectedTopicsWithSerde() {
final String applicationId = "ktable-ktable-joinOnForeignKey";
final Properties streamsConfig = mkProperties(mkMap(
- mkEntry(StreamsConfig.APPLICATION_ID_CONFIG, applicationId),
- mkEntry(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "asdf:0000"),
- mkEntry(StreamsConfig.STATE_DIR_CONFIG,
TestUtils.tempDirectory().getPath())
+ mkEntry(StreamsConfig.APPLICATION_ID_CONFIG, applicationId),
+ mkEntry(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "asdf:0000"),
+ mkEntry(StreamsConfig.STATE_DIR_CONFIG,
TestUtils.tempDirectory().getPath())
));
final UniqueTopicSerdeScope serdeScope = new UniqueTopicSerdeScope();
final StreamsBuilder builder = new StreamsBuilder();
- final KTable<String, String> left = builder.table(
- LEFT_TABLE,
- Consumed.with(serdeScope.decorateSerde(Serdes.String(),
streamsConfig, true),
- serdeScope.decorateSerde(Serdes.String(),
streamsConfig, false))
+ final KTable<Integer, String> left = builder.table(
+ LEFT_TABLE,
+ Consumed.with(serdeScope.decorateSerde(Serdes.Integer(),
streamsConfig, true),
+ serdeScope.decorateSerde(Serdes.String(),
streamsConfig, false))
);
- final KTable<String, String> right = builder.table(
- RIGHT_TABLE,
- Consumed.with(serdeScope.decorateSerde(Serdes.String(),
streamsConfig, true),
- serdeScope.decorateSerde(Serdes.String(),
streamsConfig, false))
+ final KTable<Integer, String> right = builder.table(
+ RIGHT_TABLE,
+ Consumed.with(serdeScope.decorateSerde(Serdes.Integer(),
streamsConfig, true),
+ serdeScope.decorateSerde(Serdes.String(),
streamsConfig, false))
);
left.join(
- right,
- value -> value.split("\\|")[1],
- (value1, value2) -> "(" + value1 + "," + value2 + ")",
- Materialized.with(null, serdeScope.decorateSerde(Serdes.String(),
streamsConfig, false)
- ))
- .toStream()
- .to(OUTPUT);
+ right,
+ value -> Integer.parseInt(value.split("\\|")[1]),
+ (value1, value2) -> "(" + value1 + "," + value2 + ")",
+ Materialized.with(null,
serdeScope.decorateSerde(Serdes.String(), streamsConfig, false)
+ ))
+ .toStream()
+ .to(OUTPUT);
final Topology topology = builder.build(streamsConfig);
try (final TopologyTestDriver driver = new
TopologyTestDriver(topology, streamsConfig)) {
- final TestInputTopic<String, String> leftInput =
driver.createInputTopic(LEFT_TABLE, new StringSerializer(), new
StringSerializer());
- final TestInputTopic<String, String> rightInput =
driver.createInputTopic(RIGHT_TABLE, new StringSerializer(), new
StringSerializer());
- leftInput.pipeInput("lhs1", "lhsValue1|rhs1");
- rightInput.pipeInput("rhs1", "rhsValue1");
+ final TestInputTopic<Integer, String> leftInput =
driver.createInputTopic(LEFT_TABLE, new IntegerSerializer(), new
StringSerializer());
+ final TestInputTopic<Integer, String> rightInput =
driver.createInputTopic(RIGHT_TABLE, new IntegerSerializer(), new
StringSerializer());
+ leftInput.pipeInput(2, "lhsValue1|1");
+ rightInput.pipeInput(1, "rhsValue1");
}
// verifying primarily that no extra pseudo-topics were used, but it's
nice to also verify the rest of the
// topics our serdes serialize data for
assertThat(serdeScope.registeredTopics(), is(mkSet(
- // expected pseudo-topics
- applicationId +
"-KTABLE-FK-JOIN-SUBSCRIPTION-REGISTRATION-0000000006-topic-fk--key",
- applicationId +
"-KTABLE-FK-JOIN-SUBSCRIPTION-REGISTRATION-0000000006-topic-pk--key",
- applicationId +
"-KTABLE-FK-JOIN-SUBSCRIPTION-REGISTRATION-0000000006-topic-vh--value",
- // internal topics
- applicationId +
"-KTABLE-FK-JOIN-SUBSCRIPTION-REGISTRATION-0000000006-topic--key",
- applicationId +
"-KTABLE-FK-JOIN-SUBSCRIPTION-RESPONSE-0000000014-topic--key",
- applicationId +
"-KTABLE-FK-JOIN-SUBSCRIPTION-RESPONSE-0000000014-topic--value",
- applicationId +
"-left_table-STATE-STORE-0000000000-changelog--key",
- applicationId +
"-left_table-STATE-STORE-0000000000-changelog--value",
- applicationId +
"-right_table-STATE-STORE-0000000003-changelog--key",
- applicationId +
"-right_table-STATE-STORE-0000000003-changelog--value",
- // output topics
- "output-topic--key",
- "output-topic--value"
+ // expected pseudo-topics
+ applicationId +
"-KTABLE-FK-JOIN-SUBSCRIPTION-REGISTRATION-0000000006-topic-fk--key",
+ applicationId +
"-KTABLE-FK-JOIN-SUBSCRIPTION-REGISTRATION-0000000006-topic-pk--key",
+ applicationId +
"-KTABLE-FK-JOIN-SUBSCRIPTION-REGISTRATION-0000000006-topic-vh--value",
+ // internal topics
+ applicationId +
"-KTABLE-FK-JOIN-SUBSCRIPTION-REGISTRATION-0000000006-topic--key",
+ applicationId +
"-KTABLE-FK-JOIN-SUBSCRIPTION-RESPONSE-0000000014-topic--key",
+ applicationId +
"-KTABLE-FK-JOIN-SUBSCRIPTION-RESPONSE-0000000014-topic--value",
+ applicationId +
"-left_table-STATE-STORE-0000000000-changelog--key",
+ applicationId +
"-left_table-STATE-STORE-0000000000-changelog--value",
+ applicationId +
"-right_table-STATE-STORE-0000000003-changelog--key",
+ applicationId +
"-right_table-STATE-STORE-0000000003-changelog--value",
+ // output topics
+ "output-topic--key",
+ "output-topic--value"
Review comment:
Right, sorry, my bad.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
