vvcephei commented on a change in pull request #8764:
URL: https://github.com/apache/kafka/pull/8764#discussion_r436858939
##########
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:
It'd be better to avoid reformatting code to change indentation
settings. AFAIK, in Streams at least, we tend to use indents of 4 spaces.
##########
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:
Might be nice for demonstration purposes if the two records actually
have different keys. Maybe:
```suggestion
aTopic.pipeInput(1, "999-alpha");
bTopic.pipeInput(999, "beta");
```
##########
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:
bump (it looks like this didn't get changed)
----------------------------------------------------------------
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
