Der Test verwendet die `@Bean` von `EndlessConsumer`

diff --git a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
index 6bde5ff..653e81e 100644 (file)
--- a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
+++ b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
@@ -18,7 +18,7 @@ import java.util.concurrent.Executors;
 public class ApplicationConfiguration
 {
   @Bean
-  public RecordHandler<String, Long> recordHandler()
+  public RecordHandler<String, Long> applicationRecordHandler()
   {
     return (record) ->
     {

diff --git a/src/test/java/de/juplo/kafka/ApplicationTests.java b/src/test/java/de/juplo/kafka/ApplicationTests.java
index 1781b1d..b7f8308 100644 (file)
--- a/src/test/java/de/juplo/kafka/ApplicationTests.java
+++ b/src/test/java/de/juplo/kafka/ApplicationTests.java
@@ -71,7 +71,7 @@ public class ApplicationTests extends GenericApplicationTests<String, Long>
   public static class Configuration
   {
     @Bean
-    public RecordHandler<String, Long> recordHandler()
+    public RecordHandler<String, Long> applicationRecordHandler()
     {
       return (record) ->
       {

diff --git a/src/test/java/de/juplo/kafka/GenericApplicationTests.java b/src/test/java/de/juplo/kafka/GenericApplicationTests.java
index 9465ce6..4883f75 100644 (file)
--- a/src/test/java/de/juplo/kafka/GenericApplicationTests.java
+++ b/src/test/java/de/juplo/kafka/GenericApplicationTests.java
@@ -1,7 +1,6 @@
 package de.juplo.kafka;
 
 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;
@@ -14,6 +13,7 @@ import org.junit.jupiter.api.*;
 import org.springframework.beans.factory.annotation.Autowired;
 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 +21,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;
@@ -54,18 +53,14 @@ abstract class GenericApplicationTests<K, V>
        @Autowired
        ApplicationProperties applicationProperties;
        @Autowired
-       ExecutorService executor;
+       TestRecordHandler<K, V> recordHandler;
        @Autowired
-       ConsumerRebalanceListener rebalanceListener;
-       @Autowired
-       RecordHandler<K, V> recordHandler;
+       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 +84,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 +92,7 @@ abstract class GenericApplicationTests<K, V>
                                .untilAsserted(() ->
                                {
                                        checkSeenOffsetsForProgress();
-                                       assertSeenOffsetsEqualCommittedOffsets(seenOffsets);
+                                       assertSeenOffsetsEqualCommittedOffsets(recordHandler.seenOffsets);
                                });
 
                assertThatExceptionOfType(IllegalStateException.class)
@@ -121,7 +116,7 @@ abstract class GenericApplicationTests<K, V>
                                .until(() -> !endlessConsumer.running());
 
                checkSeenOffsetsForProgress();
-               assertSeenOffsetsEqualCommittedOffsets(seenOffsets);
+               assertSeenOffsetsEqualCommittedOffsets(recordHandler.seenOffsets);
 
                endlessConsumer.start();
                await("Consumer failed")
@@ -130,8 +125,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 +153,7 @@ abstract class GenericApplicationTests<K, V>
                                .until(() -> !endlessConsumer.running());
 
                checkSeenOffsetsForProgress();
-               assertSeenOffsetsAreBehindCommittedOffsets(seenOffsets);
+               assertSeenOffsetsAreBehindCommittedOffsets(recordHandler.seenOffsets);
 
                endlessConsumer.start();
                await("Consumer failed")
@@ -166,7 +161,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 +214,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);
@@ -336,43 +331,30 @@ abstract class GenericApplicationTests<K, V>
                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 +371,10 @@ abstract class GenericApplicationTests<K, V>
        @Import(ApplicationConfiguration.class)
        public static class Configuration
        {
+               @Bean
+               public RecordHandler recordHandler(RecordHandler applicationRecordHandler)
+               {
+                       return new TestRecordHandler(applicationRecordHandler);
+               }
        }
 }

diff --git a/src/test/java/de/juplo/kafka/TestRecordHandler.java b/src/test/java/de/juplo/kafka/TestRecordHandler.java
index b4efdd6..37d3f65 100644 (file)
--- a/src/test/java/de/juplo/kafka/TestRecordHandler.java
+++ b/src/test/java/de/juplo/kafka/TestRecordHandler.java
@@ -2,16 +2,28 @@ package de.juplo.kafka;
 
 import lombok.RequiredArgsConstructor;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.TopicPartition;
+
+import java.util.Map;
+import java.util.Set;
 
 
 @RequiredArgsConstructor
-public abstract class TestRecordHandler<K, V> implements RecordHandler<K, V>
+public class TestRecordHandler<K, V> implements RecordHandler<K, V>
 {
   private final RecordHandler<K, V> handler;
 
+  Map<TopicPartition, Long> seenOffsets;
+  Set<ConsumerRecord<K, V>> receivedRecords;
 
-  public abstract void onNewRecord(ConsumerRecord<K, V> record);
 
+  public void onNewRecord(ConsumerRecord<K, V> record)
+  {
+    seenOffsets.put(
+      new TopicPartition(record.topic(), record.partition()),
+      record.offset());
+    receivedRecords.add(record);
+  }
 
   @Override
   public void accept(ConsumerRecord<K, V> record)