@RequiredArgsConstructor
static class Consumer
{
- private final List<KeyValue<Word, WordCount>> received = new LinkedList<>();
+ private final List<KeyValue<Word, WordCounter>> received = new LinkedList<>();
@KafkaListener(groupId = "TEST", topics = TOPIC_OUT)
- public synchronized void receive(ConsumerRecord<Word, WordCount> record)
+ public synchronized void receive(ConsumerRecord<Word, WordCounter> record)
{
log.debug("Received message: {}", record);
received.add(KeyValue.pair(record.key(),record.value()));
}
- synchronized List<KeyValue<Word, WordCount>> getReceivedMessages()
+ synchronized List<KeyValue<Word, WordCounter>> getReceivedMessages()
{
return received;
}
(JsonSerializer<String>)keySerde.serializer(),
(JsonSerializer<Word>)valueSerde.serializer());
- TestOutputTopic<Word, WordCount> out = testDriver.createOutputTopic(
+ TestOutputTopic<Word, WordCounter> out = testDriver.createOutputTopic(
OUT,
(JsonDeserializer<Word>)keySerde.deserializer(),
- (JsonDeserializer<WordCount>)valueSerde.deserializer());
+ (JsonDeserializer<WordCounter>)valueSerde.deserializer());
TestData.writeInputData((key, value) -> in.pipeInput(key, value));
- List<KeyValue<Word,WordCount>> receivedMessages = out
+ List<KeyValue<Word, WordCounter>> receivedMessages = out
.readRecordsToList()
.stream()
.map(record -> KeyValue.pair(record.key(), record.value()))
Word.of("klaus","s"));
}
- static void assertExpectedResult(List<KeyValue<Word, WordCount>> receivedMessages)
+ static void assertExpectedResult(List<KeyValue<Word, WordCounter>> receivedMessages)
{
assertThat(receivedMessages).hasSize(11);
assertThat(receivedMessages).containsSubsequence(
expectedMessages[9]); // Boäh
}
- static KeyValue<Word,WordCount>[] expectedMessages = new KeyValue[]
+ static KeyValue<Word, WordCounter>[] expectedMessages = new KeyValue[]
{
KeyValue.pair(
Word.of("peter","Hallo"),
- WordCount.of("peter","Hallo",1)),
+ WordCounter.of("peter","Hallo",1)),
KeyValue.pair(
Word.of("klaus","Müsch"),
- WordCount.of("klaus","Müsch",1)),
+ WordCounter.of("klaus","Müsch",1)),
KeyValue.pair(
Word.of("peter","Welt"),
- WordCount.of("peter","Welt",1)),
+ WordCounter.of("peter","Welt",1)),
KeyValue.pair(
Word.of("klaus","Müsch"),
- WordCount.of("klaus","Müsch",2)),
+ WordCounter.of("klaus","Müsch",2)),
KeyValue.pair(
Word.of("klaus","s"),
- WordCount.of("klaus","s",1)),
+ WordCounter.of("klaus","s",1)),
KeyValue.pair(
Word.of("peter","Boäh"),
- WordCount.of("peter","Boäh",1)),
+ WordCounter.of("peter","Boäh",1)),
KeyValue.pair(
Word.of("peter","Welt"),
- WordCount.of("peter","Welt",2)),
+ WordCounter.of("peter","Welt",2)),
KeyValue.pair(
Word.of("peter","Boäh"),
- WordCount.of("peter","Boäh",2)),
+ WordCounter.of("peter","Boäh",2)),
KeyValue.pair(
Word.of("klaus","s"),
- WordCount.of("klaus","s",2)),
+ WordCounter.of("klaus","s",2)),
KeyValue.pair(
Word.of("peter","Boäh"),
- WordCount.of("peter","Boäh",3)),
+ WordCounter.of("peter","Boäh",3)),
KeyValue.pair(
Word.of("klaus","s"),
- WordCount.of("klaus","s",3)),
+ WordCounter.of("klaus","s",3)),
};
static Map<String, Object> convertToMap(Properties properties)