From: Kai Moritz <kai@juplo.de>
Date: Sun, 26 Feb 2023 14:48:25 +0000 (+0100)
Subject: WIP
X-Git-Tag: kafkadata~38
X-Git-Url: https://juplo.de/gitweb/?a=commitdiff_plain;h=faeec2b848a9e4a25fed7ca1e771068ea8c59583;p=demos%2Fkafka%2Fchat

WIP
---

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);
     });
   }