X-Git-Url: http://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Ftest%2Fjava%2Fde%2Fjuplo%2Fkafka%2FApplicationTests.java;h=35d2b2e03d5510c74b784bfc334f66fff8daa1c8;hb=dac4556048bca9bfc505a0c348ec9a46313c8cf1;hp=cbf215ef4f47b97f89b46abee37aa58b87195c66;hpb=a9200a876060edc8683dfd6d0d16c23407c189ad;p=demos%2Fkafka%2Ftraining

diff --git a/src/test/java/de/juplo/kafka/ApplicationTests.java b/src/test/java/de/juplo/kafka/ApplicationTests.java
index cbf215e..35d2b2e 100644
--- a/src/test/java/de/juplo/kafka/ApplicationTests.java
+++ b/src/test/java/de/juplo/kafka/ApplicationTests.java
@@ -67,6 +67,8 @@ class ApplicationTests
 	@Autowired
 	ApplicationProperties properties;
 	@Autowired
+	EndlessConsumer endlessConsumer;
+	@Autowired
 	RecordHandler recordHandler;
 
 	Map<TopicPartition, Long> oldOffsets;
@@ -224,8 +226,22 @@ class ApplicationTests
 					newOffsets.put(
 							new TopicPartition(record.topic(), record.partition()),
 							record.offset());
+
+		endlessConsumer.start();
 	}
 
+	@AfterEach
+	public void deinit()
+	{
+		try
+		{
+			endlessConsumer.stop();
+		}
+		catch (Exception e)
+		{
+			log.info("Exception while stopping the consumer: {}", e.toString());
+		}
+	}
 
 	public static class RecordHandler implements Consumer<ConsumerRecord<String, Long>>
 	{