X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Ftest%2Fjava%2Fde%2Fjuplo%2Fkafka%2Fwordcount%2Fcounter%2FCounterApplicationIT.java;h=559b1716c80c5f54641f70fe792aaacd4a015f0a;hb=f9106e5166ce84caa2a910aa21fb41efaebcc73f;hp=75f0988aa10427b6e0525ad89a69934fd06b2c8e;hpb=0a7e20af38fe1502c7c4328b2b6024cfe9e7b019;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 75f0988..559b171 100644 --- a/src/test/java/de/juplo/kafka/wordcount/counter/CounterApplicationIT.java +++ b/src/test/java/de/juplo/kafka/wordcount/counter/CounterApplicationIT.java @@ -1,9 +1,9 @@ package de.juplo.kafka.wordcount.counter; -import com.fasterxml.jackson.databind.ObjectMapper; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier; import org.apache.kafka.streams.state.Stores; import org.junit.jupiter.api.BeforeEach; @@ -14,12 +14,18 @@ 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.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.core.ProducerFactory; +import org.springframework.kafka.support.serializer.JsonSerializer; import org.springframework.kafka.test.context.EmbeddedKafka; import java.time.Duration; import java.util.LinkedList; import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.stream.Collectors; import static de.juplo.kafka.wordcount.counter.CounterApplicationIT.*; import static org.awaitility.Awaitility.*; @@ -42,9 +48,7 @@ public class CounterApplicationIT static final int PARTITIONS = 2; @Autowired - KafkaTemplate kafkaTemplate; - @Autowired - ObjectMapper mapper; + KafkaTemplate kafkaTemplate; @Autowired Consumer consumer; @@ -63,7 +67,7 @@ public class CounterApplicationIT await("Expexted converted data") .atMost(Duration.ofSeconds(10)) - .untilAsserted(() -> TestData.assertExpectedResult(consumer.received, mapper)); + .untilAsserted(() -> TestData.assertExpectedResult(consumer.getReceivedMessages())); } @@ -73,16 +77,43 @@ public class CounterApplicationIT private final List received = new LinkedList<>(); @KafkaListener(groupId = "TEST", topics = TOPIC_OUT) - public void receive(ConsumerRecord record) + public synchronized void receive(ConsumerRecord record) { log.debug("Received message: {}", record); received.add(Message.of(record.key(),record.value())); } + + synchronized List getReceivedMessages() + { + return received; + } } @TestConfiguration static class Configuration { + @Bean + ProducerFactory producerFactory(Properties streamProcessorProperties) + { + Map propertyMap = streamProcessorProperties + .entrySet() + .stream() + .collect( + Collectors.toMap( + entry -> (String)entry.getKey(), + entry -> entry.getValue() + )); + + propertyMap.put( + ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, + JsonSerializer.class.getName()); + propertyMap.put( + ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, + JsonSerializer.class.getName()); + + return new DefaultKafkaProducerFactory<>(propertyMap); + } + @Bean Consumer consumer() {