Die App führt die Berechnung aus

[demos/kafka/training] / src / main / java / de / juplo / kafka / SimpleConsumer.java

diff --git a/src/main/java/de/juplo/kafka/SimpleConsumer.java b/src/main/java/de/juplo/kafka/SimpleConsumer.java
index 45f9b94..cea9568 100644 (file)
--- a/src/main/java/de/juplo/kafka/SimpleConsumer.java
+++ b/src/main/java/de/juplo/kafka/SimpleConsumer.java
@@ -19,6 +19,7 @@ public class SimpleConsumer implements Callable<Integer>
   private final String id;
   private final String topic;
   private final Consumer<String, Message> consumer;
+  private final MessageHandler messageHandler;
 
   private long consumed = 0;
 
@@ -76,5 +77,6 @@ public class SimpleConsumer implements Callable<Integer>
   {
     consumed++;
     log.info("{} - {}: {}/{} - {}={}", id, offset, topic, partition, key, value);
+    messageHandler.handle(key, value);
   }
 }