From e1a190e45699bcc64dad173228dce558705ad434 Mon Sep 17 00:00:00 2001
From: Kai Moritz <kai@juplo.de>
Date: Thu, 23 Jan 2025 19:48:46 +0100
Subject: [PATCH] =?utf8?q?TODO:=20So=20klappt=20es=20zwar,=20aber=20der=20?=
 =?utf8?q?Test=20f=C3=BChrt=20main()=20nicht=20aus...?=
MIME-Version: 1.0
Content-Type: text/plain; charset=utf8
Content-Transfer-Encoding: 8bit

---
 build.gradle                                  |  3 +-
 pom.xml                                       |  6 +-
 .../java/de/juplo/kafka/ExampleProducer.java  |  2 +-
 src/main/resources/application.yml            | 13 ++++
 .../java/de/juplo/kafka/ApplicationTests.java | 77 ++++++++++++++++++-
 5 files changed, 95 insertions(+), 6 deletions(-)

diff --git a/build.gradle b/build.gradle
index 7bbe8704..7dcc4f25 100644
--- a/build.gradle
+++ b/build.gradle
@@ -27,7 +27,8 @@ repositories {
 }
 
 dependencies {
-	implementation 'org.springframework.boot:spring-boot-starter'
+	implementation 'org.springframework.boot:spring-boot-starter-web'
+	implementation 'org.springframework.boot:spring-boot-starter-actuator'
 	implementation 'org.springframework.kafka:spring-kafka'
 	compileOnly 'org.projectlombok:lombok'
 	developmentOnly 'org.springframework.boot:spring-boot-devtools'
diff --git a/pom.xml b/pom.xml
index 1f08dfee..ac03bdc3 100644
--- a/pom.xml
+++ b/pom.xml
@@ -24,7 +24,11 @@
   <dependencies>
     <dependency>
       <groupId>org.springframework.boot</groupId>
-      <artifactId>spring-boot-starter</artifactId>
+      <artifactId>spring-boot-starter-web</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-actuator</artifactId>
     </dependency>
     <dependency>
       <groupId>org.springframework.boot</groupId>
diff --git a/src/main/java/de/juplo/kafka/ExampleProducer.java b/src/main/java/de/juplo/kafka/ExampleProducer.java
index 30846331..a967254f 100644
--- a/src/main/java/de/juplo/kafka/ExampleProducer.java
+++ b/src/main/java/de/juplo/kafka/ExampleProducer.java
@@ -39,7 +39,7 @@ public class ExampleProducer
           result.getRecordMetadata().offset()));
 
       completableFuture.exceptionally(e -> {
-        log.error("ERROR sendig message", e);
+        log.error("ERROR sending message", e);
         return null;
       });
       // end::callback[]
diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml
index 51cc46c7..ab6c4be4 100644
--- a/src/main/resources/application.yml
+++ b/src/main/resources/application.yml
@@ -3,3 +3,16 @@ spring:
     bootstrap-servers: :9092
     template:
       default-topic: test
+management:
+  endpoint:
+    shutdown:
+      enabled: true
+  endpoints:
+    web:
+      exposure:
+        include: "*"
+  info:
+    env:
+      enabled: true
+    java:
+      enabled: true
diff --git a/src/test/java/de/juplo/kafka/ApplicationTests.java b/src/test/java/de/juplo/kafka/ApplicationTests.java
index 714175e4..3e56a841 100644
--- a/src/test/java/de/juplo/kafka/ApplicationTests.java
+++ b/src/test/java/de/juplo/kafka/ApplicationTests.java
@@ -1,24 +1,95 @@
 package de.juplo.kafka;
 
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;
 import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.context.TestConfiguration;
+import org.springframework.context.annotation.Bean;
+import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.kafka.test.context.EmbeddedKafka;
+import org.springframework.test.web.servlet.MockMvc;
 
+import java.time.Duration;
+import java.util.LinkedList;
+import java.util.List;
+
+import static de.juplo.kafka.ApplicationTests.PARTITIONS;
 import static de.juplo.kafka.ApplicationTests.TOPIC;
+import static org.awaitility.Awaitility.await;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
 
 
 @SpringBootTest(
     properties = {
         "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
-        "spring.kafka.template.default-topic=" + TOPIC
+        "spring.kafka.consumer.auto-offset-reset=earliest",
     })
-@EmbeddedKafka(topics = TOPIC)
+@AutoConfigureMockMvc
+@EmbeddedKafka(topics = TOPIC, partitions = PARTITIONS)
+@Slf4j
 public class ApplicationTests
 {
-  public final static String TOPIC = "out";
+  static final String TOPIC = "FOO";
+  static final int PARTITIONS = 10;
+
+  @Autowired
+  MockMvc mockMvc;
+  @Autowired
+  Consumer consumer;
+
+
+  @BeforeEach
+  public void clear()
+  {
+    consumer.received.clear();
+  }
+
 
   @Test
   public void testApplicationStartup()
   {
+    await("Application is healthy")
+      .atMost(Duration.ofSeconds(5))
+      .untilAsserted(() -> mockMvc
+        .perform(get("/actuator/health"))
+        .andExpect(status().isOk())
+        .andExpect(jsonPath("status").value("UP")));
+  }
+
+  @Test
+  public void testSendMessage() throws Exception
+  {
+    await("Some messages were send")
+        .atMost(Duration.ofSeconds(5))
+        .until(() -> consumer.received.size() >= 1);
+  }
+
+
+  static class Consumer
+  {
+    final List<ConsumerRecord<String, String>> received = new LinkedList<>();
+
+    @KafkaListener(groupId = "TEST", topics = TOPIC)
+    public void receive(ConsumerRecord<String, String> record)
+    {
+      log.debug("Received message: {}", record);
+      received.add(record);
+    }
+  }
+
+  @TestConfiguration
+  static class Configuration
+  {
+    @Bean
+    Consumer consumer()
+    {
+      return new Consumer();
+    }
   }
 }
-- 
2.20.1