Der Value-Typ in dem Topic `state` ist jetzt auch vom Typ `String`

* Dadurch wird die Kontrolle der Ergebnisse einfacher, da alle Nachrichten
  auch einfach mit `kafkacat` gelesen werden können.

diff --git a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
index c1fe03a..e454ed0 100644 (file)
--- 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 9e743cd..7584a17 100644 (file)
--- 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)