X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2FApplicationRecordHandler.java;fp=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2FApplicationRecordHandler.java;h=e5e40eda4627fdedb15b440fd768532f842b838f;hb=dd9103fdedc432ba861abeaf5cfb4acb66749f15;hp=51d524fc682beb6a8998075a55bf037c7c25ee51;hpb=0c89ddec8156e2f44fe28c9d05fe06f548e9168e;p=demos%2Fkafka%2Ftraining

diff --git a/src/main/java/de/juplo/kafka/ApplicationRecordHandler.java b/src/main/java/de/juplo/kafka/ApplicationRecordHandler.java
index 51d524f..e5e40ed 100644
--- a/src/main/java/de/juplo/kafka/ApplicationRecordHandler.java
+++ b/src/main/java/de/juplo/kafka/ApplicationRecordHandler.java
@@ -28,6 +28,9 @@ public class ApplicationRecordHandler implements RecordHandler<String, String>
     String user = record.key();
     String message = record.value();
 
+    if (!state.containsKey(partition))
+      state.put(partition, new AdderBusinessLogic());
+
     if (message.equals("CALCULATE"))
     {
       AdderResult result = state.get(partition).calculate(user);