Merge branch 'deserialization' into sumup-adder--ohne--stored-offsets

[demos/kafka/training] / src / test / java / de / juplo / kafka / GenericApplicationTests.java

diff --git a/src/test/java/de/juplo/kafka/GenericApplicationTests.java b/src/test/java/de/juplo/kafka/GenericApplicationTests.java
index 9465ce6..e16aea7 100644 (file)
--- a/src/test/java/de/juplo/kafka/GenericApplicationTests.java
+++ b/src/test/java/de/juplo/kafka/GenericApplicationTests.java
@@ -1,7 +1,7 @@
 package de.juplo.kafka;
 
+import com.mongodb.client.MongoClient;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.clients.producer.KafkaProducer;
@@ -12,8 +12,12 @@ import org.apache.kafka.common.serialization.*;
 import org.apache.kafka.common.utils.Bytes;
 import org.junit.jupiter.api.*;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.autoconfigure.mongo.MongoProperties;
+import org.springframework.boot.test.autoconfigure.data.mongo.AutoConfigureDataMongo;
 import org.springframework.boot.test.context.ConfigDataApplicationContextInitializer;
 import org.springframework.boot.test.context.TestConfiguration;
+import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Import;
 import org.springframework.kafka.test.context.EmbeddedKafka;
 import org.springframework.test.context.TestPropertySource;
@@ -21,7 +25,6 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 
 import java.time.Duration;
 import java.util.*;
-import java.util.concurrent.ExecutorService;
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
@@ -36,10 +39,13 @@ import static org.awaitility.Awaitility.*;
 @SpringJUnitConfig(initializers = ConfigDataApplicationContextInitializer.class)
 @TestPropertySource(
                properties = {
-                               "consumer.bootstrap-server=${spring.embedded.kafka.brokers}",
-                               "consumer.topic=" + TOPIC,
-                               "consumer.commit-interval=500ms" })
+                               "sumup.adder.bootstrap-server=${spring.embedded.kafka.brokers}",
+                               "sumup.adder.topic=" + TOPIC,
+                               "sumup.adder.commit-interval=500ms",
+                               "spring.mongodb.embedded.version=4.4.13" })
 @EmbeddedKafka(topics = TOPIC, partitions = PARTITIONS)
+@EnableAutoConfiguration
+@AutoConfigureDataMongo
 @Slf4j
 abstract class GenericApplicationTests<K, V>
 {
@@ -54,18 +60,19 @@ abstract class GenericApplicationTests<K, V>
        @Autowired
        ApplicationProperties applicationProperties;
        @Autowired
-       ExecutorService executor;
+       MongoClient mongoClient;
        @Autowired
-       ConsumerRebalanceListener rebalanceListener;
+       MongoProperties mongoProperties;
        @Autowired
-       RecordHandler<K, V> recordHandler;
+       RebalanceListener rebalanceListener;
+       @Autowired
+       TestRecordHandler<K, V> recordHandler;
+       @Autowired
+       EndlessConsumer<K, V> endlessConsumer;
 
        KafkaProducer<Bytes, Bytes> testRecordProducer;
        KafkaConsumer<Bytes, Bytes> offsetConsumer;
-       EndlessConsumer<K, V> endlessConsumer;
        Map<TopicPartition, Long> oldOffsets;
-       Map<TopicPartition, Long> seenOffsets;
-       Set<ConsumerRecord<K, V>> receivedRecords;
 
 
        final RecordGenerator recordGenerator;
@@ -89,7 +96,7 @@ abstract class GenericApplicationTests<K, V>
                await(numberOfGeneratedMessages + " records received")
                                .atMost(Duration.ofSeconds(30))
                                .pollInterval(Duration.ofSeconds(1))
-                               .until(() -> receivedRecords.size() >= numberOfGeneratedMessages);
+                               .until(() -> recordHandler.receivedRecords.size() >= numberOfGeneratedMessages);
 
                await("Offsets committed")
                                .atMost(Duration.ofSeconds(10))
@@ -97,7 +104,7 @@ abstract class GenericApplicationTests<K, V>
                                .untilAsserted(() ->
                                {
                                        checkSeenOffsetsForProgress();
-                                       assertSeenOffsetsEqualCommittedOffsets(seenOffsets);
+                                       assertSeenOffsetsEqualCommittedOffsets(recordHandler.seenOffsets);
                                });
 
                assertThatExceptionOfType(IllegalStateException.class)
@@ -121,7 +128,7 @@ abstract class GenericApplicationTests<K, V>
                                .until(() -> !endlessConsumer.running());
 
                checkSeenOffsetsForProgress();
-               assertSeenOffsetsEqualCommittedOffsets(seenOffsets);
+               assertSeenOffsetsEqualCommittedOffsets(recordHandler.seenOffsets);
 
                endlessConsumer.start();
                await("Consumer failed")
@@ -130,8 +137,8 @@ abstract class GenericApplicationTests<K, V>
                                .until(() -> !endlessConsumer.running());
 
                checkSeenOffsetsForProgress();
-               assertSeenOffsetsEqualCommittedOffsets(seenOffsets);
-               assertThat(receivedRecords.size())
+               assertSeenOffsetsEqualCommittedOffsets(recordHandler.seenOffsets);
+               assertThat(recordHandler.receivedRecords.size())
                                .describedAs("Received not all sent events")
                                .isLessThan(numberOfGeneratedMessages);
 
@@ -158,7 +165,7 @@ abstract class GenericApplicationTests<K, V>
                                .until(() -> !endlessConsumer.running());
 
                checkSeenOffsetsForProgress();
-               assertSeenOffsetsAreBehindCommittedOffsets(seenOffsets);
+               assertSeenOffsetsAreBehindCommittedOffsets(recordHandler.seenOffsets);
 
                endlessConsumer.start();
                await("Consumer failed")
@@ -166,7 +173,7 @@ abstract class GenericApplicationTests<K, V>
                                .pollInterval(Duration.ofSeconds(1))
                                .until(() -> !endlessConsumer.running());
 
-               assertSeenOffsetsAreBehindCommittedOffsets(seenOffsets);
+               assertSeenOffsetsAreBehindCommittedOffsets(recordHandler.seenOffsets);
 
                assertThatNoException()
                                .describedAs("Consumer should not be running")
@@ -219,7 +226,7 @@ abstract class GenericApplicationTests<K, V>
                partitions().forEach(tp ->
                {
                        Long oldOffset = oldOffsets.get(tp) + 1;
-                       Long newOffset = seenOffsets.get(tp) + 1;
+                       Long newOffset = recordHandler.seenOffsets.get(tp) + 1;
                        if (!oldOffset.equals(newOffset))
                        {
                                log.debug("Progress for {}: {} -> {}", tp, oldOffset, newOffset);
@@ -333,46 +340,34 @@ abstract class GenericApplicationTests<K, V>
                props.put("value.deserializer", BytesDeserializer.class.getName());
                offsetConsumer = new KafkaConsumer<>(props);
 
+               mongoClient.getDatabase(mongoProperties.getDatabase()).drop();
                seekToEnd();
 
                oldOffsets = new HashMap<>();
-               seenOffsets = new HashMap<>();
-               receivedRecords = new HashSet<>();
+               recordHandler.seenOffsets = new HashMap<>();
+               recordHandler.receivedRecords = new HashSet<>();
 
                doForCurrentOffsets((tp, offset) ->
                {
                        oldOffsets.put(tp, offset - 1);
-                       seenOffsets.put(tp, offset - 1);
+                       recordHandler.seenOffsets.put(tp, offset - 1);
                });
 
-               TestRecordHandler<K, V> captureOffsetAndExecuteTestHandler =
-                               new TestRecordHandler<K, V>(recordHandler)
-                               {
-                                       @Override
-                                       public void onNewRecord(ConsumerRecord<K, V> record)
-                                       {
-                                               seenOffsets.put(
-                                                               new TopicPartition(record.topic(), record.partition()),
-                                                               record.offset());
-                                               receivedRecords.add(record);
-                                               consumer.accept(record);
-                                       }
-                               };
-
-               endlessConsumer =
-                               new EndlessConsumer<>(
-                                               executor,
-                                               applicationProperties.getClientId(),
-                                               applicationProperties.getTopic(),
-                                               kafkaConsumer,
-                                               captureOffsetAndExecuteTestHandler);
-
                endlessConsumer.start();
        }
 
        @AfterEach
        public void deinit()
        {
+               try
+               {
+                       endlessConsumer.stop();
+               }
+               catch (Exception e)
+               {
+                       log.debug("{}", e.toString());
+               }
+
                try
                {
                        testRecordProducer.close();
@@ -389,5 +384,10 @@ abstract class GenericApplicationTests<K, V>
        @Import(ApplicationConfiguration.class)
        public static class Configuration
        {
+               @Bean
+               public RecordHandler recordHandler(RecordHandler applicationRecordHandler)
+               {
+                       return new TestRecordHandler(applicationRecordHandler);
+               }
        }
 }