From 522c21f5323272fafdda57b61ff52f05e23dd521 Mon Sep 17 00:00:00 2001
From: Kai Moritz <kai@juplo.de>
Date: Thu, 19 Dec 2024 21:34:17 +0100
Subject: [PATCH] =?utf8?q?Konfiguration=20=C3=BCber=20`ApplicationConigura?=
 =?utf8?q?tion`=20-=20aber=20von=20Hand?=
MIME-Version: 1.0
Content-Type: text/plain; charset=utf8
Content-Transfer-Encoding: 8bit

---
 .../de/juplo/kafka/ExampleConsumerTest.java   | 20 +++++--------------
 1 file changed, 5 insertions(+), 15 deletions(-)

diff --git a/src/test/java/de/juplo/kafka/ExampleConsumerTest.java b/src/test/java/de/juplo/kafka/ExampleConsumerTest.java
index 03269fd2..590c9cdf 100644
--- a/src/test/java/de/juplo/kafka/ExampleConsumerTest.java
+++ b/src/test/java/de/juplo/kafka/ExampleConsumerTest.java
@@ -4,12 +4,9 @@ import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.admin.AdminClientConfig;
 import org.apache.kafka.clients.admin.RecordsToDelete;
 import org.apache.kafka.clients.consumer.Consumer;
-import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.clients.producer.RecordMetadata;
 import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.common.serialization.LongDeserializer;
 import org.apache.kafka.common.serialization.LongSerializer;
-import org.apache.kafka.common.serialization.StringDeserializer;
 import org.awaitility.Awaitility;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
@@ -25,10 +22,8 @@ import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.test.context.EmbeddedKafka;
 
 import java.time.Duration;
-import java.util.Collection;
 import java.util.HashMap;
 import java.util.Map;
-import java.util.Properties;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
@@ -41,9 +36,11 @@ import static de.juplo.kafka.ExampleConsumerTest.TOPIC;
 @SpringBootTest(
   classes = {
     KafkaAutoConfiguration.class,
+    ApplicationProperties.class,
     ExampleConsumerTest.ConsumerRunnableTestConfig.class,
   },
   properties = {
+    "juplo.bootstrap-server=${spring.embedded.kafka.brokers}",
     "spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.ByteArraySerializer",
     "logging.level.de.juplo.kafka=TRACE",
   })
@@ -164,17 +161,10 @@ public class ExampleConsumerTest
 
 
   @BeforeEach
-  void createExampleConsumer(@Value("${spring.embedded.kafka.brokers}") String kafkaBroker)
+  void createExampleConsumer(@Autowired ApplicationProperties properties)
   {
-    Properties props = new Properties();
-    props.put("bootstrap.servers", kafkaBroker);
-    props.put("client.id", ID);
-    props.put("group.id", ID);
-    props.put("auto.offset.reset", "earliest");
-    props.put("key.deserializer", StringDeserializer.class.getName());
-    props.put("value.deserializer", LongDeserializer.class.getName());
-
-    Consumer<String, Long> consumer = new KafkaConsumer<>(props);
+    ApplicationConfiguration configuration = new ApplicationConfiguration();
+    Consumer<String, Long> consumer = configuration.kafkaConsumer(properties);
 
     exampleConsumer = new ExampleConsumer(
       ID,
-- 
2.20.1