X-Git-Url: http://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Ftest%2Fjava%2Fde%2Fjuplo%2Fkafka%2FGenericApplicationTests.java;h=753debe2b00344766c2c9a4f90286e0b6e075337;hb=f095f71a104fcde025a63f87ba75eb5cb3136656;hp=21c3f7f01c524eb64ad88d204b9b7a82e4a65e22;hpb=7adff476ad862d30d668d75212d1ca1c7cf16b03;p=demos%2Fkafka%2Ftraining

diff --git a/src/test/java/de/juplo/kafka/GenericApplicationTests.java b/src/test/java/de/juplo/kafka/GenericApplicationTests.java
index 21c3f7f..753debe 100644
--- a/src/test/java/de/juplo/kafka/GenericApplicationTests.java
+++ b/src/test/java/de/juplo/kafka/GenericApplicationTests.java
@@ -2,7 +2,6 @@ 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;
@@ -14,20 +13,21 @@ 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.kafka.KafkaAutoConfiguration;
 import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
 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.config.KafkaListenerEndpointRegistry;
+import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.test.context.EmbeddedKafka;
 import org.springframework.test.context.TestPropertySource;
 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;
@@ -39,11 +39,7 @@ import static org.assertj.core.api.Assertions.*;
 import static org.awaitility.Awaitility.*;
 
 
-@SpringJUnitConfig(
-  initializers = ConfigDataApplicationContextInitializer.class,
-  classes = {
-      KafkaAutoConfiguration.class,
-      ApplicationTests.Configuration.class })
+@SpringJUnitConfig(initializers = ConfigDataApplicationContextInitializer.class)
 @TestPropertySource(
 		properties = {
 				"spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
@@ -63,28 +59,23 @@ abstract class GenericApplicationTests<K, V>
 	@Autowired
 	org.apache.kafka.clients.consumer.Consumer<K, V> kafkaConsumer;
 	@Autowired
-	Consumer<ConsumerRecord<K, V>> consumer;
-	@Autowired
-	ApplicationProperties applicationProperties;
-	@Autowired
 	KafkaProperties kafkaProperties;
 	@Autowired
-	ExecutorService executor;
+	ApplicationProperties applicationProperties;
 	@Autowired
 	MongoClient mongoClient;
 	@Autowired
 	MongoProperties mongoProperties;
 	@Autowired
-	ConsumerRebalanceListener rebalanceListener;
+	KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;
+	@Autowired
+	TestRecordHandler<K, V> recordHandler;
 	@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;
@@ -108,7 +99,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))
@@ -116,7 +107,7 @@ abstract class GenericApplicationTests<K, V>
 				.untilAsserted(() ->
 				{
 					checkSeenOffsetsForProgress();
-					assertSeenOffsetsEqualCommittedOffsets(seenOffsets);
+					assertSeenOffsetsEqualCommittedOffsets(recordHandler.seenOffsets);
 				});
 
 		assertThatExceptionOfType(IllegalStateException.class)
@@ -140,7 +131,7 @@ abstract class GenericApplicationTests<K, V>
 				.until(() -> !endlessConsumer.running());
 
 		checkSeenOffsetsForProgress();
-		assertSeenOffsetsEqualCommittedOffsets(seenOffsets);
+		assertSeenOffsetsEqualCommittedOffsets(recordHandler.seenOffsets);
 
 		endlessConsumer.start();
 		await("Consumer failed")
@@ -149,8 +140,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);
 
@@ -177,7 +168,7 @@ abstract class GenericApplicationTests<K, V>
 				.until(() -> !endlessConsumer.running());
 
 		checkSeenOffsetsForProgress();
-		assertSeenOffsetsAreBehindCommittedOffsets(seenOffsets);
+		assertSeenOffsetsAreBehindCommittedOffsets(recordHandler.seenOffsets);
 
 		endlessConsumer.start();
 		await("Consumer failed")
@@ -185,7 +176,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")
@@ -238,7 +229,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);
@@ -356,43 +347,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);
-					}
-				};
-
-		endlessConsumer =
-				new EndlessConsumer<>(
-						executor,
-						kafkaProperties.getClientId(),
-						applicationProperties.getTopic(),
-						kafkaConsumer,
-						rebalanceListener,
-						captureOffsetAndExecuteTestHandler);
-
 		endlessConsumer.start();
 	}
 
 	@AfterEach
 	public void deinit()
 	{
+		try
+		{
+			endlessConsumer.stop();
+		}
+		catch (Exception e)
+		{
+			log.debug("{}", e.toString());
+		}
+
 		try
 		{
 			testRecordProducer.close();
@@ -409,5 +387,16 @@ abstract class GenericApplicationTests<K, V>
 	@Import(ApplicationConfiguration.class)
 	public static class Configuration
 	{
+		@Bean
+		public RecordHandler recordHandler(RecordHandler applicationRecordHandler)
+		{
+			return new TestRecordHandler(applicationRecordHandler);
+		}
+
+    @Bean(destroyMethod = "close")
+    public org.apache.kafka.clients.consumer.Consumer<String, Message> kafkaConsumer(ConsumerFactory<String, Message> factory)
+    {
+      return factory.createConsumer();
+    }
 	}
 }