From f7475320b20be8ba198ba914958e9e4dddf62e11 Mon Sep 17 00:00:00 2001
From: Kai Moritz <kai@juplo.de>
Date: Sun, 26 Feb 2023 19:04:38 +0100
Subject: [PATCH] WIP

---
 .../persistence/kafka/ChatHomeLoader.java     | 19 +++++++++++++------
 .../kafka/KafkaChatHomeService.java           | 19 +++++++++++--------
 2 files changed, 24 insertions(+), 14 deletions(-)

diff --git a/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/ChatHomeLoader.java b/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/ChatHomeLoader.java
index 365bb5e2..15d968a7 100644
--- a/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/ChatHomeLoader.java
+++ b/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/ChatHomeLoader.java
@@ -18,7 +18,7 @@ import java.util.UUID;
 @Slf4j
 class ChatHomeLoader
 {
-  private final long offsetOfFirstNewMessage;
+  private final long offsetOfFirstUnseenMessage;
   private final ZoneId zoneId;
   private final Map<UUID, KafkaChatRoomService> kafkaChatRoomServiceMap = new HashMap<>();
 
@@ -33,10 +33,19 @@ class ChatHomeLoader
    */
   boolean handleMessage(ConsumerRecord<UUID, MessageTo> record)
   {
-    if (record.offset() >= offsetOfFirstNewMessage)
+    Message.MessageKey messageKey = Message.MessageKey.of(
+        record.value().getUser(),
+        record.value().getId());
+
+    if (record.offset() >= offsetOfFirstUnseenMessage)
     {
       // All messages consumed: DONE!
-      log.debug("I");
+      log.trace(
+          "Ignoring unseen message {}: topic={}, partition={}, offset={}",
+          messageKey,
+          record.topic(),
+          record.partition(),
+          record.offset());
       return true;
     }
 
@@ -49,9 +58,7 @@ class ChatHomeLoader
         });
 
     service.addMessage(new Message(
-        Message.MessageKey.of(
-            record.value().getUser(),
-            record.value().getId()),
+        messageKey,
         record.offset(),
         time,
         record.value().getText()
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 a5d63fd8..e171bc57 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,8 @@ public class KafkaChatHomeService implements ChatHomeService, ConsumerRebalanceL
   private final Consumer<String, MessageTo> consumer;
   private final String topic;
   // private final long[] offsets; Erst mal immer alles neu einlesen
-  private final Map<UUID, ChatRoom>[] chatrooms;
+  private final Map<UUID, KafkaChatRoomService>[] kafkaChatRoomServiceMaps;
+  private final Map<UUID, ChatRoom>[] chatRoomMaps;
 
 
   public KafkaChatHomeService(
@@ -34,7 +35,8 @@ public class KafkaChatHomeService implements ChatHomeService, ConsumerRebalanceL
     // {
     //   this.offsets[i] = 0l;
     // }
-    this.chatrooms = new Map[numShards];
+    this.kafkaChatRoomServiceMaps = new Map[numShards];
+    this.chatRoomMaps = new Map[numShards];
   }
 
 
@@ -50,6 +52,7 @@ public class KafkaChatHomeService implements ChatHomeService, ConsumerRebalanceL
       }
 
       int partition = tp.partition();
+      kafkaChatRoomServiceMaps[partition] = new HashMap<>(); // TODO: reuse! Nicht immer alles neu laden
       long unseenOffset = 0; // offsets[partition];
 
       log.info(
@@ -59,7 +62,7 @@ public class KafkaChatHomeService implements ChatHomeService, ConsumerRebalanceL
           currentOffset);
 
       consumer.seek(tp, unseenOffset);
-      chatrooms[partition]
+      chatRoomMaps[partition]
           .values()
           .stream()
       handlers[partition] = new ChatRoomLoadingMessageHandlingStrategy(tp, currentOffset, unseenOffset);
@@ -101,7 +104,7 @@ public class KafkaChatHomeService implements ChatHomeService, ConsumerRebalanceL
         (a, b) -> a.addAll(b));
     for (int shard = 0; shard < numShards; shard++)
     {
-      chatrooms[shard] = owned.contains(shard)
+      chatRoomMaps[shard] = owned.contains(shard)
         ? new HashMap<>()
         : null;
     }
@@ -119,25 +122,25 @@ public class KafkaChatHomeService implements ChatHomeService, ConsumerRebalanceL
         }
       })
       .toStream()
-      .forEach(chatroom -> chatrooms[chatroom.getShard()].put(chatroom.getId(), chatroom));
+      .forEach(chatroom -> chatRoomMaps[chatroom.getShard()].put(chatroom.getId(), chatroom));
   }
 
   @Override
   public Mono<ChatRoom> putChatRoom(ChatRoom chatRoom)
   {
-    chatrooms[chatRoom.getShard()].put(chatRoom.getId(), chatRoom);
+    chatRoomMaps[chatRoom.getShard()].put(chatRoom.getId(), chatRoom);
     return Mono.just(chatRoom);
   }
 
   @Override
   public Mono<ChatRoom> getChatRoom(int shard, UUID id)
   {
-    return Mono.justOrEmpty(chatrooms[shard].get(id));
+    return Mono.justOrEmpty(chatRoomMaps[shard].get(id));
   }
 
   @Override
   public Flux<ChatRoom> getChatRooms(int shard)
   {
-    return Flux.fromStream(chatrooms[shard].values().stream());
+    return Flux.fromStream(chatRoomMaps[shard].values().stream());
   }
 }
-- 
2.20.1