</parent>
<groupId>de.juplo.kafka.wordcount</groupId>
<artifactId>top10</artifactId>
- <version>1.1.1</version>
+ <version>1.1.2</version>
<name>Wordcount-Top-10</name>
<description>Top-10 stream-processor of the multi-user wordcount-example</description>
<properties>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
+
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.springframework.kafka</groupId>
+ <artifactId>spring-kafka-test</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.awaitility</groupId>
+ <artifactId>awaitility</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.assertj</groupId>
+ <artifactId>assertj-core</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
<plugins>
+ <plugin>
+ <artifactId>maven-failsafe-plugin</artifactId>
+ </plugin>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
--- /dev/null
+package de.juplo.kafka.wordcount.top10;
+
+
+import lombok.AccessLevel;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+
+@NoArgsConstructor
+@AllArgsConstructor(
+ staticName = "of",
+ access = AccessLevel.PACKAGE)
+@Data
+public class Counter
+{
+ String user;
+ String word;
+ long counter;
+}
--- /dev/null
+package de.juplo.kafka.wordcount.top10;
+
+import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.streams.KeyValue;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.function.BiConsumer;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+
+class TestData
+{
+ static void writeInputData(BiConsumer<Key, Counter> consumer)
+ {
+ consumer.accept(
+ Key.of("peter","Hallo"),
+ Counter.of("peter","Hallo",1));
+ consumer.accept(
+ Key.of("klaus","Müsch"),
+ Counter.of("klaus","Müsch",1));
+ consumer.accept(
+ Key.of("peter","Welt"),
+ Counter.of("peter","Welt",1));
+ consumer.accept(
+ Key.of("klaus","Müsch"),
+ Counter.of("klaus","Müsch",2));
+ consumer.accept(
+ Key.of("klaus","s"),
+ Counter.of("klaus","s",1));
+ consumer.accept(
+ Key.of("peter","Boäh"),
+ Counter.of("peter","Boäh",1));
+ consumer.accept(
+ Key.of("peter","Welt"),
+ Counter.of("peter","Welt",2));
+ consumer.accept(
+ Key.of("peter","Boäh"),
+ Counter.of("peter","Boäh",2));
+ consumer.accept(
+ Key.of("klaus","s"),
+ Counter.of("klaus","s",2));
+ consumer.accept(
+ Key.of("peter","Boäh"),
+ Counter.of("peter","Boäh",3));
+ consumer.accept(
+ Key.of("klaus","s"),
+ Counter.of("klaus","s",3));
+ }
+
+ static void assertExpectedResult(List<KeyValue<String, Ranking>> receivedMessages)
+ {
+ assertThat(receivedMessages).hasSize(11);
+ assertThat(receivedMessages).containsSubsequence(
+ expectedMessages[0],
+ expectedMessages[2],
+ expectedMessages[5],
+ expectedMessages[6],
+ expectedMessages[7]); // peter
+
+ assertThat(receivedMessages).containsSubsequence(
+ expectedMessages[1],
+ expectedMessages[3],
+ expectedMessages[4],
+ expectedMessages[8],
+ expectedMessages[10]); // klaus
+ }
+
+ static KeyValue<String, Ranking>[] expectedMessages = 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 Map<String, Object> 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());
+ }
+ }
+}
--- /dev/null
+package de.juplo.kafka.wordcount.top10;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.streams.*;
+import org.apache.kafka.streams.state.Stores;
+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 java.time.Instant;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.function.BiConsumer;
+
+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<String, Object> 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<Key, Entry> in = testDriver.createInputTopic(
+ IN,
+ (JsonSerializer<Key>)keySerde.serializer(),
+ (JsonSerializer<Entry>)valueSerde.serializer());
+
+ TestOutputTopic<String, Ranking> out = testDriver.createOutputTopic(
+ OUT,
+ (JsonDeserializer<String>)keySerde.deserializer(),
+ (JsonDeserializer<Ranking>)valueSerde.deserializer());
+
+ TestData.writeInputData(new BiConsumer<Key, Counter>()
+ {
+ private Instant timestamp = Instant.now();
+
+ @Override
+ public void accept(Key key, Counter value)
+ {
+ in.pipeInput(
+ key,
+ Entry.of(value.getWord(), value.getCounter()),
+ timestamp);
+ timestamp = timestamp.plusMillis(500);
+ }
+ });
+
+ List<KeyValue<String, Ranking>> receivedMessages = out
+ .readRecordsToList()
+ .stream()
+ .map(record ->
+ {
+ log.debug(
+ "OUT: {} -> {}, {}, {}",
+ record.key(),
+ record.value(),
+ parseHeader(record.headers(), KEY_DEFAULT_CLASSID_FIELD_NAME),
+ parseHeader(record.headers(), DEFAULT_CLASSID_FIELD_NAME));
+ return KeyValue.pair(record.key(), record.value());
+ })
+ .toList();
+
+ TestData.assertExpectedResult(receivedMessages);
+
+ testDriver.close();
+ }
+}
projects / demos / kafka / wordcount / commitdiff