From 21b1d0cffa0bfe47d7d78cef4e4dc23588d69a1b Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Sat, 25 May 2024 16:59:42 +0200 Subject: [PATCH] top10: 1.1.2 - (RED) Added test, that asserts the expectated processing --- pom.xml | 21 ++- .../kafka/wordcount/counter/TestCounter.java | 21 +++ .../kafka/wordcount/counter/TestWord.java | 17 ++ .../juplo/kafka/wordcount/top10/TestData.java | 159 ++++++++++++++++++ .../Top10StreamProcessorTopologyTest.java | 83 +++++++++ 5 files changed, 300 insertions(+), 1 deletion(-) create mode 100644 src/test/java/de/juplo/kafka/wordcount/counter/TestCounter.java create mode 100644 src/test/java/de/juplo/kafka/wordcount/counter/TestWord.java create mode 100644 src/test/java/de/juplo/kafka/wordcount/top10/TestData.java create mode 100644 src/test/java/de/juplo/kafka/wordcount/top10/Top10StreamProcessorTopologyTest.java diff --git a/pom.xml b/pom.xml index 6749738..86b4290 100644 --- a/pom.xml +++ b/pom.xml @@ -10,7 +10,7 @@ de.juplo.kafka.wordcount top10 - 1.1.1 + 1.1.2 Wordcount-Top-10 Top-10 stream-processor of the multi-user wordcount-example @@ -50,15 +50,34 @@ lombok true + org.springframework.boot spring-boot-starter-test test + + org.springframework.kafka + spring-kafka-test + test + + + org.awaitility + awaitility + test + + + org.assertj + assertj-core + test + + + maven-failsafe-plugin + org.springframework.boot spring-boot-maven-plugin diff --git a/src/test/java/de/juplo/kafka/wordcount/counter/TestCounter.java b/src/test/java/de/juplo/kafka/wordcount/counter/TestCounter.java new file mode 100644 index 0000000..d98ae64 --- /dev/null +++ b/src/test/java/de/juplo/kafka/wordcount/counter/TestCounter.java @@ -0,0 +1,21 @@ +package de.juplo.kafka.wordcount.counter; + +import lombok.AllArgsConstructor; +import lombok.Data; +import lombok.NoArgsConstructor; + + +@Data +@NoArgsConstructor +@AllArgsConstructor(staticName = "of") +public class TestCounter +{ + String user; + String word; + long counter; + + public static TestCounter of(TestWord word, long counter) + { + return new TestCounter(word.getUser(), word.getWord(), counter); + } +} diff --git a/src/test/java/de/juplo/kafka/wordcount/counter/TestWord.java b/src/test/java/de/juplo/kafka/wordcount/counter/TestWord.java new file mode 100644 index 0000000..8008e12 --- /dev/null +++ b/src/test/java/de/juplo/kafka/wordcount/counter/TestWord.java @@ -0,0 +1,17 @@ +package de.juplo.kafka.wordcount.counter; + +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import lombok.AllArgsConstructor; +import lombok.Data; +import lombok.NoArgsConstructor; + + +@AllArgsConstructor(staticName = "of") +@NoArgsConstructor +@Data +@JsonIgnoreProperties(ignoreUnknown = true) +public class TestWord +{ + private String user; + private String word; +} diff --git a/src/test/java/de/juplo/kafka/wordcount/top10/TestData.java b/src/test/java/de/juplo/kafka/wordcount/top10/TestData.java new file mode 100644 index 0000000..73a405e --- /dev/null +++ b/src/test/java/de/juplo/kafka/wordcount/top10/TestData.java @@ -0,0 +1,159 @@ +package de.juplo.kafka.wordcount.top10; + +import de.juplo.kafka.wordcount.counter.TestCounter; +import de.juplo.kafka.wordcount.counter.TestWord; +import org.apache.kafka.common.header.Header; +import org.apache.kafka.common.header.Headers; +import org.apache.kafka.streams.KeyValue; +import org.springframework.util.LinkedMultiValueMap; +import org.springframework.util.MultiValueMap; + +import java.util.Map; +import java.util.Properties; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +import static org.assertj.core.api.Assertions.assertThat; + + +class TestData +{ + static final KeyValue[] INPUT_MESSAGES = new KeyValue[] + { + new KeyValue<>( + TestWord.of("peter","Hallo"), + TestCounter.of("peter","Hallo",1)), + new KeyValue<>( + TestWord.of("klaus","Müsch"), + TestCounter.of("klaus","Müsch",1)), + new KeyValue<>( + TestWord.of("peter","Welt"), + TestCounter.of("peter","Welt",1)), + new KeyValue<>( + TestWord.of("klaus","Müsch"), + TestCounter.of("klaus","Müsch",2)), + new KeyValue<>( + TestWord.of("klaus","s"), + TestCounter.of("klaus","s",1)), + new KeyValue<>( + TestWord.of("peter","Boäh"), + TestCounter.of("peter","Boäh",1)), + new KeyValue<>( + TestWord.of("peter","Welt"), + TestCounter.of("peter","Welt",2)), + new KeyValue<>( + TestWord.of("peter","Boäh"), + TestCounter.of("peter","Boäh",2)), + new KeyValue<>( + TestWord.of("klaus","s"), + TestCounter.of("klaus","s",2)), + new KeyValue<>( + TestWord.of("peter","Boäh"), + TestCounter.of("peter","Boäh",3)), + new KeyValue<>( + TestWord.of("klaus","s"), + TestCounter.of("klaus","s",3)), + }; + + static void assertExpectedMessages(MultiValueMap receivedMessages) + { + expectedMessages().forEach( + (user, rankings) -> + assertThat(receivedMessages.get(user)) + .containsExactlyElementsOf(rankings)); + } + + static KeyValue[] EXPECTED_MESSAGES = new KeyValue[] + { + KeyValue.pair( // 0 + "peter", + Ranking.of( + Entry.of("Hallo", 1l))), + KeyValue.pair( // 1 + "klaus", + Ranking.of( + Entry.of("Müsch", 1l))), + KeyValue.pair( // 2 + "peter", + Ranking.of( + Entry.of("Hallo", 1l), + Entry.of("Welt", 1l))), + KeyValue.pair( // 3 + "klaus", + Ranking.of( + Entry.of("Müsch", 2l))), + KeyValue.pair( // 4 + "klaus", + Ranking.of( + Entry.of("Müsch", 2l), + Entry.of("s", 1l))), + KeyValue.pair( // 5 + "peter", + Ranking.of( + Entry.of("Hallo", 1l), + Entry.of("Welt", 1l), + Entry.of("Boäh", 1l))), + KeyValue.pair( // 6 + "peter", + Ranking.of( + Entry.of("Welt", 2l), + Entry.of("Hallo", 1l), + Entry.of("Boäh", 1l))), + KeyValue.pair( // 7 + "peter", + Ranking.of( + Entry.of("Welt", 2l), + Entry.of("Boäh", 2l), + Entry.of("Hallo", 1l))), + KeyValue.pair( // 8 + "klaus", + Ranking.of( + Entry.of("Müsch", 2l), + Entry.of("s", 2l))), + KeyValue.pair( // 9 + "peter", + Ranking.of( + Entry.of("Boäh", 3l), + Entry.of("Welt", 2l), + Entry.of("Hallo", 1l))), + KeyValue.pair( // 10 + "klaus", + Ranking.of( + Entry.of("s", 3l), + Entry.of("Müsch", 2l))), + }; + + static MultiValueMap expectedMessages() + { + MultiValueMap expectedMessages = new LinkedMultiValueMap<>(); + Stream + .of(EXPECTED_MESSAGES) + .forEach(keyValue -> expectedMessages.add(keyValue.key, keyValue.value)); + return expectedMessages; + } + + static Map convertToMap(Properties properties) + { + return properties + .entrySet() + .stream() + .collect( + Collectors.toMap( + entry -> (String)entry.getKey(), + entry -> entry.getValue() + )); + } + + static String parseHeader(Headers headers, String key) + { + Header header = headers.lastHeader(key); + if (header == null) + { + return key + "=null"; + } + else + { + return key + "=" + new String(header.value()); + } + } +} diff --git a/src/test/java/de/juplo/kafka/wordcount/top10/Top10StreamProcessorTopologyTest.java b/src/test/java/de/juplo/kafka/wordcount/top10/Top10StreamProcessorTopologyTest.java new file mode 100644 index 0000000..8ecf9fa --- /dev/null +++ b/src/test/java/de/juplo/kafka/wordcount/top10/Top10StreamProcessorTopologyTest.java @@ -0,0 +1,83 @@ +package de.juplo.kafka.wordcount.top10; + +import lombok.extern.slf4j.Slf4j; +import org.apache.kafka.streams.TestInputTopic; +import org.apache.kafka.streams.TestOutputTopic; +import org.apache.kafka.streams.Topology; +import org.apache.kafka.streams.TopologyTestDriver; +import org.junit.jupiter.api.Test; +import org.springframework.kafka.support.serializer.JsonDeserializer; +import org.springframework.kafka.support.serializer.JsonSerde; +import org.springframework.kafka.support.serializer.JsonSerializer; +import org.springframework.util.LinkedMultiValueMap; +import org.springframework.util.MultiValueMap; + +import java.util.Map; +import java.util.Properties; +import java.util.stream.Stream; + +import static de.juplo.kafka.wordcount.top10.TestData.convertToMap; +import static de.juplo.kafka.wordcount.top10.TestData.parseHeader; +import static org.springframework.kafka.support.mapping.AbstractJavaTypeMapper.DEFAULT_CLASSID_FIELD_NAME; +import static org.springframework.kafka.support.mapping.AbstractJavaTypeMapper.KEY_DEFAULT_CLASSID_FIELD_NAME; + + +@Slf4j +public class Top10StreamProcessorTopologyTest +{ + public final static String IN = "TEST-IN"; + public final static String OUT = "TEST-OUT"; + + @Test + public void test() + { + Topology topology = Top10StreamProcessor.buildTopology(IN, OUT); + + Top10ApplicationConfiguration applicationConfiguriation = + new Top10ApplicationConfiguration(); + Properties streamProcessorProperties = + applicationConfiguriation.streamProcessorProperties(new Top10ApplicationProperties()); + Map propertyMap = convertToMap(streamProcessorProperties); + + JsonSerde keySerde = new JsonSerde<>(); + keySerde.configure(propertyMap, true); + JsonSerde valueSerde = new JsonSerde<>(); + valueSerde.configure(propertyMap, false); + + TopologyTestDriver testDriver = new TopologyTestDriver(topology, streamProcessorProperties); + + TestInputTopic in = testDriver.createInputTopic( + IN, + (JsonSerializer)keySerde.serializer(), + (JsonSerializer)valueSerde.serializer()); + + TestOutputTopic out = testDriver.createOutputTopic( + OUT, + (JsonDeserializer)keySerde.deserializer(), + (JsonDeserializer)valueSerde.deserializer()); + + Stream + .of(TestData.INPUT_MESSAGES) + .forEach(kv -> in.pipeInput( + Key.of(kv.key.getUser(), kv.key.getWord()), + Entry.of(kv.value.getWord(), kv.value.getCounter()))); + + MultiValueMap receivedMessages = new LinkedMultiValueMap<>(); + out + .readRecordsToList() + .forEach(record -> + { + log.debug( + "OUT: {} -> {}, {}, {}", + record.key(), + record.value(), + parseHeader(record.headers(), KEY_DEFAULT_CLASSID_FIELD_NAME), + parseHeader(record.headers(), DEFAULT_CLASSID_FIELD_NAME)); + receivedMessages.add(record.key(), record.value()); + }); + + TestData.assertExpectedMessages(receivedMessages); + + testDriver.close(); + } +} -- 2.20.1