From c6d1ded087826ef257b54c69fb7c654b21d55a81 Mon Sep 17 00:00:00 2001
From: Kai Moritz <kai@juplo.de>
Date: Mon, 28 Oct 2024 08:00:48 +0100
Subject: [PATCH] Der Value-Typ in dem Topic `state` ist jetzt auch vom Typ
 `String`
MIME-Version: 1.0
Content-Type: text/plain; charset=utf8
Content-Transfer-Encoding: 8bit

* Dadurch wird die Kontrolle der Ergebnisse einfacher, da alle Nachrichten
  auch einfach mit `kafkacat` gelesen werden können.
---
 src/main/java/de/juplo/kafka/ApplicationConfiguration.java | 7 +++----
 src/main/java/de/juplo/kafka/ExampleConsumer.java          | 6 +++---
 2 files changed, 6 insertions(+), 7 deletions(-)

diff --git a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
index c1fe03a..e454ed0 100644
--- a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
+++ b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
@@ -5,7 +5,6 @@ import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.clients.consumer.StickyAssignor;
 import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.Producer;
-import org.apache.kafka.common.serialization.LongSerializer;
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.apache.kafka.common.serialization.StringSerializer;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
@@ -23,7 +22,7 @@ public class ApplicationConfiguration
   @Bean
   public ExampleConsumer exampleConsumer(
       Consumer<String, String> kafkaConsumer,
-      Producer<String, Long> kafkaProducer,
+      Producer<String, String> kafkaProducer,
       ApplicationProperties properties,
       ConfigurableApplicationContext applicationContext)
   {
@@ -61,7 +60,7 @@ public class ApplicationConfiguration
   }
 
   @Bean
-  public KafkaProducer<String, Long> kafkaProducer(ApplicationProperties properties)
+  public KafkaProducer<String, String> kafkaProducer(ApplicationProperties properties)
   {
     Properties props = new Properties();
     props.put("bootstrap.servers", properties.getBootstrapServer());
@@ -74,7 +73,7 @@ public class ApplicationConfiguration
     props.put("linger.ms", properties.getProducerProperties().getLingerMs());
     props.put("compression.type", properties.getProducerProperties().getCompressionType());
     props.put("key.serializer", StringSerializer.class.getName());
-    props.put("value.serializer", LongSerializer.class.getName());
+    props.put("value.serializer", StringSerializer.class.getName());
 
     return new KafkaProducer<>(props);
   }
diff --git a/src/main/java/de/juplo/kafka/ExampleConsumer.java b/src/main/java/de/juplo/kafka/ExampleConsumer.java
index da845bd..aa489c4 100644
--- a/src/main/java/de/juplo/kafka/ExampleConsumer.java
+++ b/src/main/java/de/juplo/kafka/ExampleConsumer.java
@@ -29,7 +29,7 @@ public class ExampleConsumer implements Runnable, ConsumerRebalanceListener
 
   private final Map<String, Long> counterState = new HashMap<>();
   private final String stateTopic;
-  private final Producer<String, Long> producer;
+  private final Producer<String, String> producer;
 
   private volatile boolean running = false;
   private final Phaser phaser = new Phaser(1);
@@ -44,7 +44,7 @@ public class ExampleConsumer implements Runnable, ConsumerRebalanceListener
     String topic,
     Consumer<String, String> consumer,
     String stateTopic,
-    Producer<String, Long> producer,
+    Producer<String, String> producer,
     Runnable closeCallback)
   {
     this.id = clientId;
@@ -153,7 +153,7 @@ public class ExampleConsumer implements Runnable, ConsumerRebalanceListener
   void sendCounterState(int partition, String key, Long counter)
   {
     seen[partition]++;
-    ProducerRecord<String, Long> record = new ProducerRecord<>(stateTopic, key, counter);
+    ProducerRecord<String, String> record = new ProducerRecord<>(stateTopic, key, counter.toString());
     producer.send(record, ((metadata, exception) ->
     {
       if (exception == null)
-- 
2.20.1