Version des `spring-producer`, der JSON-Nachrichten sendet

diff --git a/README.sh b/README.sh
index 499780a..68deb85 100755 (executable)
--- a/README.sh
+++ b/README.sh
@@ -1,6 +1,6 @@
 #!/bin/bash
 
-IMAGE=juplo/spring-producer:1.0-SNAPSHOT
+IMAGE=juplo/spring-producer:1.0-json-SNAPSHOT
 
 if [ "$1" = "cleanup" ]
 then

diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml
index c417a7f..24a6ab6 100644 (file)
--- a/docker/docker-compose.yml
+++ b/docker/docker-compose.yml
@@ -190,19 +190,25 @@ services:
       - kafka-3
 
   producer:
-    image: juplo/spring-producer:1.0-SNAPSHOT
+    image: juplo/spring-producer:1.0-json-SNAPSHOT
     environment:
       juplo.bootstrap-server: kafka:9092
       juplo.client-id: producer
       juplo.producer.topic: test
 
   consumer-1:
-    image: juplo/simple-consumer:1.0-SNAPSHOT
-    command: kafka:9092 test my-group consumer-1
+    image: juplo/spring-consumer:1.1-json-SNAPSHOT
+    environment:
+      juplo.bootstrap-server: kafka:9092
+      juplo.client-id: consumer-1
+      juplo.consumer.topic: test
 
   consumer-2:
-    image: juplo/simple-consumer:1.0-SNAPSHOT
-    command: kafka:9092 test my-group consumer-2
+    image: juplo/spring-consumer:1.1-json-SNAPSHOT
+    environment:
+      juplo.bootstrap-server: kafka:9092
+      juplo.client-id: consumer-2
+      juplo.consumer.topic: test
 
 volumes:
   zookeeper-data:

diff --git a/pom.xml b/pom.xml
index 841299b..88c8a33 100644 (file)
--- a/pom.xml
+++ b/pom.xml
@@ -15,7 +15,7 @@
   <artifactId>spring-producer</artifactId>
   <name>Spring Producer</name>
   <description>A Simple Spring-Boot-Producer, that takes messages via POST and confirms successs</description>
-  <version>1.0-SNAPSHOT</version>
+  <version>1.0-json-SNAPSHOT</version>
 
   <properties>
     <java.version>21</java.version>
@@ -40,8 +40,8 @@
       <artifactId>spring-boot-starter-validation</artifactId>
     </dependency>
     <dependency>
-      <groupId>org.apache.kafka</groupId>
-      <artifactId>kafka-clients</artifactId>
+      <groupId>org.springframework.kafka</groupId>
+      <artifactId>spring-kafka</artifactId>
     </dependency>
     <dependency>
       <groupId>org.projectlombok</groupId>
@@ -52,11 +52,6 @@
       <artifactId>spring-boot-starter-test</artifactId>
       <scope>test</scope>
     </dependency>
-    <dependency>
-      <groupId>org.springframework.kafka</groupId>
-      <artifactId>spring-kafka</artifactId>
-      <scope>test</scope>
-    </dependency>
     <dependency>
       <groupId>org.springframework.kafka</groupId>
       <artifactId>spring-kafka-test</artifactId>

diff --git a/src/main/java/de/juplo/kafka/AddNumberMessage.java b/src/main/java/de/juplo/kafka/AddNumberMessage.java
new file mode 100644 (file)
index 0000000..deb6350
--- /dev/null
+++ b/src/main/java/de/juplo/kafka/AddNumberMessage.java
@@ -0,0 +1,11 @@
+package de.juplo.kafka;
+
+import lombok.Value;
+
+
+@Value
+public class AddNumberMessage implements SumupMessage
+{
+  private final int number;
+  private final int next;
+}

diff --git a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
index 7540dd3..d5ce01a 100644 (file)
--- a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
+++ b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
@@ -7,6 +7,7 @@ import org.springframework.boot.context.properties.EnableConfigurationProperties
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.support.serializer.JsonSerializer;
 
 import java.time.Duration;
 import java.util.Properties;
@@ -19,7 +20,7 @@ public class ApplicationConfiguration
   @Bean
   public ExampleProducer exampleProducer(
       ApplicationProperties properties,
-      Producer<String, String> kafkaProducer,
+      Producer<String, SumupMessage> kafkaProducer,
       ConfigurableApplicationContext applicationContext)
   {
     return
@@ -34,7 +35,7 @@ public class ApplicationConfiguration
   }
 
   @Bean(destroyMethod = "")
-  public KafkaProducer<String, String> kafkaProducer(ApplicationProperties properties)
+  public KafkaProducer<String, SumupMessage> kafkaProducer(ApplicationProperties properties)
   {
     Properties props = new Properties();
     props.put("bootstrap.servers", properties.getBootstrapServer());
@@ -49,7 +50,8 @@ public class ApplicationConfiguration
     props.put("linger.ms", properties.getProducerProperties().getLinger().toMillis());
     props.put("compression.type", properties.getProducerProperties().getCompressionType());
     props.put("key.serializer", StringSerializer.class.getName());
-    props.put("value.serializer", StringSerializer.class.getName());
+    props.put("value.serializer", JsonSerializer.class.getName());
+    props.put("spring.json.type.mapping", "ADD:de.juplo.kafka.AddNumberMessage,CALC:de.juplo.kafka.CalculateSumMessage");
 
     return new KafkaProducer<>(props);
   }

diff --git a/src/main/java/de/juplo/kafka/CalculateSumMessage.java b/src/main/java/de/juplo/kafka/CalculateSumMessage.java
new file mode 100644 (file)
index 0000000..6aa0121
--- /dev/null
+++ b/src/main/java/de/juplo/kafka/CalculateSumMessage.java
@@ -0,0 +1,11 @@
+package de.juplo.kafka;
+
+
+import lombok.Value;
+
+
+@Value
+public class CalculateSumMessage implements SumupMessage
+{
+  private final int number;
+}

diff --git a/src/main/java/de/juplo/kafka/ExampleProducer.java b/src/main/java/de/juplo/kafka/ExampleProducer.java
index bc5cf89..2360704 100644 (file)
--- a/src/main/java/de/juplo/kafka/ExampleProducer.java
+++ b/src/main/java/de/juplo/kafka/ExampleProducer.java
@@ -13,7 +13,7 @@ public class ExampleProducer implements Runnable
   private final String id;
   private final String topic;
   private final Duration throttle;
-  private final Producer<String, String> producer;
+  private final Producer<String, SumupMessage> producer;
   private final Thread workerThread;
   private final Runnable closeCallback;
 
@@ -25,7 +25,7 @@ public class ExampleProducer implements Runnable
     String id,
     String topic,
     Duration throttle,
-    Producer<String, String> producer,
+    Producer<String, SumupMessage> producer,
     Runnable closeCallback)
   {
     this.id = id;
@@ -43,13 +43,18 @@ public class ExampleProducer implements Runnable
   @Override
   public void run()
   {
-    long i = 0;
+    int i = 0;
 
     try
     {
       for (; running; i++)
       {
-        send(Long.toString(i%10), Long.toString(i));
+        int number = i % 10;
+        SumupMessage message = (i % 7 == 0)
+          ? new CalculateSumMessage(number)
+          : new AddNumberMessage(number, i);
+
+        send(Long.toString(number), message);
 
         if (throttle.isPositive())
         {
@@ -78,11 +83,11 @@ public class ExampleProducer implements Runnable
     }
   }
 
-  void send(String key, String value)
+  void send(String key, SumupMessage value)
   {
     final long time = System.currentTimeMillis();
 
-    final ProducerRecord<String, String> record = new ProducerRecord<>(
+    final ProducerRecord<String, SumupMessage> record = new ProducerRecord<>(
         topic,  // Topic
         key,    // Key
         value   // Value

diff --git a/src/main/java/de/juplo/kafka/SumupMessage.java b/src/main/java/de/juplo/kafka/SumupMessage.java
new file mode 100644 (file)
index 0000000..739efd1
--- /dev/null
+++ b/src/main/java/de/juplo/kafka/SumupMessage.java
@@ -0,0 +1,5 @@
+package de.juplo.kafka;
+
+public interface SumupMessage
+{
+}