X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Ftest%2Fjava%2Fde%2Fjuplo%2Fkafka%2Fwordcount%2Fcounter%2FCounterApplicationIT.java;h=cb0a5b78ce7a8d00b7320e9d89f7b93531caba51;hb=236aa3a7521bf44e2e397fa7b90373cc3cf59341;hp=aad79b89e82b77393b813712fdefe850596f65b9;hpb=8a56919f493faa0acc6eb939b9c78312a1418efa;p=demos%2Fkafka%2Fwordcount diff --git a/src/test/java/de/juplo/kafka/wordcount/counter/CounterApplicationIT.java b/src/test/java/de/juplo/kafka/wordcount/counter/CounterApplicationIT.java index aad79b8..cb0a5b7 100644 --- a/src/test/java/de/juplo/kafka/wordcount/counter/CounterApplicationIT.java +++ b/src/test/java/de/juplo/kafka/wordcount/counter/CounterApplicationIT.java @@ -1,48 +1,58 @@ package de.juplo.kafka.wordcount.counter; -import com.fasterxml.jackson.core.JsonProcessingException; -import com.fasterxml.jackson.databind.ObjectMapper; -import lombok.RequiredArgsConstructor; -import lombok.Value; +import de.juplo.kafka.wordcount.splitter.TestInputWord; +import de.juplo.kafka.wordcount.top10.TestOutputWord; +import de.juplo.kafka.wordcount.top10.TestOutputWordCounter; import lombok.extern.slf4j.Slf4j; -import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier; +import org.apache.kafka.streams.state.Stores; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.boot.test.context.TestConfiguration; import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Primary; import org.springframework.kafka.annotation.KafkaListener; import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.support.KafkaHeaders; import org.springframework.kafka.test.context.EmbeddedKafka; +import org.springframework.messaging.handler.annotation.Header; +import org.springframework.messaging.handler.annotation.Payload; import org.springframework.util.LinkedMultiValueMap; import org.springframework.util.MultiValueMap; import java.time.Duration; -import static de.juplo.kafka.wordcount.counter.CounterApplicationIT.*; -import static org.assertj.core.api.Assertions.assertThat; -import static org.awaitility.Awaitility.*; +import static de.juplo.kafka.wordcount.counter.CounterApplicationIT.TOPIC_IN; +import static de.juplo.kafka.wordcount.counter.CounterApplicationIT.TOPIC_OUT; +import static org.awaitility.Awaitility.await; @SpringBootTest( properties = { - "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}", + "spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer", + "spring.kafka.producer.properties.spring.json.add.type.headers=false", + "spring.kafka.consumer.auto-offset-reset=earliest", + "spring.kafka.consumer.key-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer", + "spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer", + "spring.kafka.consumer.properties.spring.json.type.mapping=word:de.juplo.kafka.wordcount.top10.TestOutputWord,counter:de.juplo.kafka.wordcount.top10.TestOutputWordCounter", + "logging.level.root=WARN", + "logging.level.de.juplo=DEBUG", "juplo.wordcount.counter.bootstrap-server=${spring.embedded.kafka.brokers}", + "juplo.wordcount.counter.commit-interval=0", + "juplo.wordcount.counter.cacheMaxBytes=0", "juplo.wordcount.counter.input-topic=" + TOPIC_IN, "juplo.wordcount.counter.output-topic=" + TOPIC_OUT }) -@EmbeddedKafka(topics = { TOPIC_IN, TOPIC_OUT }, partitions = PARTITIONS) +@EmbeddedKafka(topics = { TOPIC_IN, TOPIC_OUT }) @Slf4j public class CounterApplicationIT { - public final static String TOPIC_IN = "in"; - public final static String TOPIC_OUT = "out"; - static final int PARTITIONS = 2; + public static final String TOPIC_IN = "in"; + public static final String TOPIC_OUT = "out"; @Autowired - KafkaTemplate kafkaTemplate; - @Autowired - ObjectMapper mapper; + KafkaTemplate kafkaTemplate; @Autowired Consumer consumer; @@ -55,95 +65,52 @@ public class CounterApplicationIT @Test - void testSendMessage() throws Exception + void testSendMessage() { - kafkaTemplate.send(TOPIC_IN, "peter", "Hallo"); - kafkaTemplate.send(TOPIC_IN, "klaus", "Müsch"); - kafkaTemplate.send(TOPIC_IN, "peter", "Welt"); - kafkaTemplate.send(TOPIC_IN, "klaus", "Müsch"); - kafkaTemplate.send(TOPIC_IN, "klaus", "s"); - kafkaTemplate.send(TOPIC_IN, "peter", "Boäh"); - kafkaTemplate.send(TOPIC_IN, "peter", "Welt"); - kafkaTemplate.send(TOPIC_IN, "peter", "Boäh"); - kafkaTemplate.send(TOPIC_IN, "klaus", "s"); - kafkaTemplate.send(TOPIC_IN, "peter", "Boäh"); - kafkaTemplate.send(TOPIC_IN, "klaus", "s"); - - Message peter1 = Message.of( - "{\"username\":\"peter\",\"word\":\"Hallo\"}", - "1"); - Message peter2 = Message.of( - "{\"username\":\"peter\",\"word\":\"Welt\"}", - "1"); - Message peter3 = Message.of( - "{\"username\":\"peter\",\"word\":\"Boäh\"}", - "1"); - Message peter4 = Message.of( - "{\"username\":\"peter\",\"word\":\"Welt\"}", - "2"); - Message peter5 = Message.of( - "{\"username\":\"peter\",\"word\":\"Boäh\"}", - "2"); - Message peter6 = Message.of( - "{\"username\":\"peter\",\"word\":\"Boäh\"}", - "3"); - - Message klaus1 = Message.of( - "{\"username\":\"klaus\",\"word\":\"Müsch\"}", - "1"); - Message klaus2 = Message.of( - "{\"username\":\"klaus\",\"word\":\"Müsch\"}", - "2"); - Message klaus3 = Message.of( - "{\"username\":\"klaus\",\"word\":\"s\"}", - "1"); - Message klaus4 = Message.of( - "{\"username\":\"klaus\",\"word\":\"s\"}", - "2"); - Message klaus5 = Message.of( - "{\"username\":\"klaus\",\"word\":\"s\"}", - "3"); - - await("Expexted converted data") + TestData + .getInputMessages() + .forEach(kv -> kafkaTemplate.send(TOPIC_IN, kv.key, kv.value)); + + await("Expected messages") .atMost(Duration.ofSeconds(10)) - .untilAsserted(() -> - { - assertThat(consumer.received).hasSize(2); - assertThat(consumer.received.get("klaus")).containsExactly(klaus1, klaus2, klaus3, klaus4, klaus5); - assertThat(consumer.received.get("peter")).containsExactly(peter1, peter2, peter3, peter4, peter5, peter6); - }); + .untilAsserted(() -> consumer.enforceAssertion(TestData.expectedMessagesAssertion())); } - @RequiredArgsConstructor static class Consumer { - private final MultiValueMap received = new LinkedMultiValueMap<>(); - private final ObjectMapper mapper; + private final MultiValueMap received = new LinkedMultiValueMap<>(); @KafkaListener(groupId = "TEST", topics = TOPIC_OUT) - public void receive(ConsumerRecord record) throws JsonProcessingException + public synchronized void receive( + @Header(KafkaHeaders.RECEIVED_KEY) TestOutputWord word, + @Payload TestOutputWordCounter counter) { - log.debug("Received message: {}", record); - Key key = mapper.readValue(record.key(), Key.class); - received.add(key.getUsername(), Message.of(record.key(),record.value())); + log.debug("Received message: {} -> {}", word, counter); + received.add(word, counter); } - } - @Value(staticConstructor = "of") - static class Message - { - String key; - String value; + synchronized void enforceAssertion( + java.util.function.Consumer> assertion) + { + assertion.accept(received); + } } @TestConfiguration static class Configuration { @Bean - Consumer consumer(ObjectMapper mapper) + Consumer consumer() + { + return new Consumer(); + } + + @Primary + @Bean + KeyValueBytesStoreSupplier inMemoryStoreSupplier() { - return new Consumer(mapper); + return Stores.inMemoryKeyValueStore("TEST-STORE"); } } }