X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2Fchat%2Fbackend%2Fpersistence%2Fkafka%2FKafkaChatHomeService.java;h=b2ed284221e6fc99bdf829b0518307297fa2a934;hb=faeec2b848a9e4a25fed7ca1e771068ea8c59583;hp=2e3b42f6f8693339481c8f1f8bbfaf5ef5223433;hpb=a5b72cd8868eb96b5282643d423d0ced8dabf069;p=demos%2Fkafka%2Fchat

diff --git a/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/KafkaChatHomeService.java b/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/KafkaChatHomeService.java
index 2e3b42f6..b2ed2842 100644
--- a/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/KafkaChatHomeService.java
+++ b/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/KafkaChatHomeService.java
@@ -18,7 +18,6 @@ public class KafkaChatHomeService implements ChatHomeService, ConsumerRebalanceL
   private final Consumer<String, MessageTo> consumer;
   private final String topic;
   private final long[] offsets;
-  private final MessageHandler[] handlers;
   private final Map<UUID, ChatRoom>[] chatrooms;
 
 
@@ -31,13 +30,14 @@ public class KafkaChatHomeService implements ChatHomeService, ConsumerRebalanceL
     this.consumer = consumer;
     this.topic = topic;
     this.offsets = new long[numShards];
-    this.handlers = new MessageHandler[numShards];
+    this.chatrooms = new Map<>[numShards];
     for (int i=0; i< numShards; i++)
     {
       this.offsets[i] = 0l;
-      this.handlers[i] = new MessageHandler(consumer, new TopicPartition(topic, i));
+      this.chatrooms[i] = chat
     }
     this.chatrooms = new Map[numShards];
+    this.chatrooms = new Map[numShards];
   }
 
 
@@ -55,7 +55,11 @@ public class KafkaChatHomeService implements ChatHomeService, ConsumerRebalanceL
       int partition = tp.partition();
       long unseenOffset = offsets[partition];
 
-      log.info("Reading partition {} from {} -> {}", partition, unseenOffset, currentOffset);
+      log.info(
+          "Loading messages from partition {}: start-offset={} -> current-offset={}",
+          partition,
+          unseenOffset,
+          currentOffset);
       handlers[partition] = new ChatRoomLoadingMessageHandlingStrategy(tp, currentOffset, unseenOffset);
     });
   }