Vorlage an `spring/spring-consumer--vorlage` angeglichen

diff --git a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
index b98c401..2b2799e 100644 (file)
--- a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
+++ b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
@@ -18,7 +18,7 @@ public class ApplicationConfiguration
 {
   @Bean
   public ExampleConsumer exampleConsumer(
-    Consumer<String, String> kafkaConsumer,
+    Consumer<String, String> kafkaConsumer, // << TODO: Typisierung anpassen
     ApplicationProperties properties,
     ConfigurableApplicationContext applicationContext)
   {
@@ -31,6 +31,7 @@ public class ApplicationConfiguration
   }
 
   @Bean(destroyMethod = "")
+  // TODO: Typisierung anpassen
   public KafkaConsumer<String, String> kafkaConsumer(ApplicationProperties properties)
   {
     Properties props = new Properties();
@@ -48,7 +49,7 @@ public class ApplicationConfiguration
     props.put("metadata.maxage.ms", 5000); //  5 Sekunden
     props.put("partition.assignment.strategy", StickyAssignor.class.getName());
     props.put("key.deserializer", StringDeserializer.class.getName());
-    props.put("value.deserializer", StringDeserializer.class.getName());
+    props.put("value.deserializer", StringDeserializer.class.getName()); // << TODO: Passende Strategie konfigurieren
 
     return new KafkaConsumer<>(props);
   }

diff --git a/src/main/java/de/juplo/kafka/ExampleConsumer.java b/src/main/java/de/juplo/kafka/ExampleConsumer.java
index 15c3959..1d25f18 100644 (file)
--- a/src/main/java/de/juplo/kafka/ExampleConsumer.java
+++ b/src/main/java/de/juplo/kafka/ExampleConsumer.java
@@ -15,7 +15,7 @@ public class ExampleConsumer implements Runnable
 {
   private final String id;
   private final String topic;
-  private final Consumer<String, String> consumer;
+  private final Consumer<String, String> consumer; // << TODO: Typisierung anpassen
   private final Thread workerThread;
   private final Runnable closeCallback;
 
@@ -45,22 +45,18 @@ public class ExampleConsumer implements Runnable
     try
     {
       log.info("{} - Subscribing to topic {}", id, topic);
-      consumer.subscribe(Arrays.asList(topic));
+      // TODO: Aufruf von consumer.subscribe()
 
       while (true)
       {
-        ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
-
-        log.info("{} - Received {} messages", id, records.count());
-        for (ConsumerRecord<String, String> record : records)
-        {
-          handleRecord(
-            record.topic(),
-            record.partition(),
-            record.offset(),
-            record.key(),
-            record.value());
-        }
+        // TODO: Poll & Consume aus Ihrer bisherigen Implementierung
+        // TODO: Verarbeitung abhängig vom Typ der Nachricht auslösen
+        // switch (value.getType())
+        // {
+        //   case ADD  -> addNumber((MessageAddNumber)value);
+        //   case CALC -> calcSum((MessageCalculateSum)value);
+        //  default   -> log.error("{} - Ignoring message of unknown typ {}", id, value.getType());
+        //}
       }
     }
     catch(WakeupException e)
@@ -82,23 +78,6 @@ public class ExampleConsumer implements Runnable
     }
   }
 
-  private void handleRecord(
-    String topic,
-    Integer partition,
-    Long offset,
-    String key,
-    String value)
-  {
-    consumed++;
-    log.info("{} - partition={}-{}, offset={}: {}={}", id, topic, partition, offset, key, value);
-    // switch (value.getType())
-    // {
-    //   case ADD  -> addNumber((MessageAddNumber)value);
-    //   case CALC -> calcSum((MessageCalculateSum)value);
-    //  default   -> log.error("{} - Ignoring message of unknown typ {}", id, value.getType());
-    //}
-  }
-
   private void addNumber(MessageAddNumber addNumber)
   {
     log.info("{} - Adding number {}", id, addNumber.getNext());