From 1fb148e961483be6aeb1170ff0673a2682f9ad01 Mon Sep 17 00:00:00 2001
From: Kai Moritz <kai@juplo.de>
Date: Sat, 15 Apr 2023 14:28:30 +0200
Subject: [PATCH] NEU

---
 .../kafka/KafkaChatHomeService.java           | 52 ++++++++-----------
 1 file changed, 23 insertions(+), 29 deletions(-)

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 f105902c..912295d6 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
@@ -29,8 +29,10 @@ public class KafkaChatHomeService implements ChatHomeService, Runnable, Consumer
   private final Producer<String, MessageTo> producer;
   private final String topic;
   private final ZoneId zoneId;
-  // private final long[] offsets; Erst mal immer alles neu einlesen
+  private final int numShards;
   private final boolean[] isShardOwned;
+  private final long[] currentOffset;
+  private final long[] nextOffset;
   private final Map<UUID, ChatRoom>[] chatRoomMaps;
   private final KafkaLikeShardingStrategy shardingStrategy;
 
@@ -52,12 +54,10 @@ public class KafkaChatHomeService implements ChatHomeService, Runnable, Consumer
     this.producer = producer;
     this.topic = topic;
     this.zoneId = zoneId;
-    // this.offsets = new long[numShards];
-    // for (int i=0; i< numShards; i++)
-    // {
-    //   this.offsets[i] = 0l;
-    // }
+    this.numShards = numShards;
     this.isShardOwned = new boolean[numShards];
+    this.currentOffset = new long[numShards];
+    this.nextOffset = new long[numShards];
     this.chatRoomMaps = new Map[numShards];
     this.shardingStrategy = new KafkaLikeShardingStrategy(numShards);
   }
@@ -70,23 +70,17 @@ public class KafkaChatHomeService implements ChatHomeService, Runnable, Consumer
 
     consumer.endOffsets(partitions).forEach((topicPartition, currentOffset) ->
     {
-      if (!topicPartition.topic().equals(topic))
-      {
-        log.warn("Ignoring partition from unwanted topic: {}", topicPartition);
-        return;
-      }
-
       int partition = topicPartition.partition();
-      long unseenOffset = 0; // offsets[partition];
+      isShardOwned[partition] =  true;
+      this.currentOffset[partition] = currentOffset;
 
       log.info(
-          "Loading messages from partition {}: start-offset={} -> current-offset={}",
+          "Partition assigned: {} - loading messages: next={} -> current={}",
           partition,
-          unseenOffset,
+          nextOffset[partition],
           currentOffset);
 
-      // TODO: reuse! Nicht immer alles neu laden, sondern erst ab offsets[partition]!
-      consumer.seek(topicPartition, unseenOffset);
+      consumer.seek(topicPartition, nextOffset[partition]);
     });
 
     consumer.resume(partitions);
@@ -97,21 +91,16 @@ public class KafkaChatHomeService implements ChatHomeService, Runnable, Consumer
   {
     partitions.forEach(topicPartition ->
     {
-      if (!topicPartition.topic().equals(topic))
-      {
-        log.warn("Ignoring partition from unwanted topic: {}", topicPartition);
-        return;
-      }
-
       int partition = topicPartition.partition();
-      // long unseenOffset = offsets[partition]; TODO: Offset merken...?
+      isShardOwned[partition] = false;
+      log.info("Partition revoked: {} - next={}", partition, nextOffset[partition]);
     });
-    log.info("Revoked partitions: {}", partitions);
   }
 
   @Override
   public void onPartitionsLost(Collection<TopicPartition> partitions)
   {
+    log.warn("Lost partitions: {}, partitions");
     // TODO: Muss auf den Verlust anders reagiert werden?
     onPartitionsRevoked(partitions);
   }
@@ -136,13 +125,18 @@ public class KafkaChatHomeService implements ChatHomeService, Runnable, Consumer
           {
             UUID chatRoomId = UUID.fromString(record.key());
             MessageTo messageTo = record.value();
-            ChatRoom chatRoom = chatRoomMaps[record.partition()].get(chatRoomId);
-            KafkaChatRoomService kafkaChatRoomService =
-                (KafkaChatRoomService) chatRoom.getChatRoomService();
+
             Message.MessageKey key = Message.MessageKey.of(messageTo.getUser(), messageTo.getId());
+
             Instant instant = Instant.ofEpochSecond(record.timestamp());
             LocalDateTime timestamp = LocalDateTime.ofInstant(instant, zoneId);
+
             Message message = new Message(key, record.offset(), timestamp, messageTo.getText());
+
+            ChatRoom chatRoom = chatRoomMaps[record.partition()].get(chatRoomId);
+            KafkaChatRoomService kafkaChatRoomService =
+                (KafkaChatRoomService) chatRoom.getChatRoomService();
+
             kafkaChatRoomService.persistMessage(message);
           }
         }
@@ -150,7 +144,7 @@ public class KafkaChatHomeService implements ChatHomeService, Runnable, Consumer
         {
           if (!records.isEmpty())
           {
-            throw new IllegalStateException("All owned partions should be paused, when no load is in progress!");
+            throw new IllegalStateException("All owned partitions should be paused, when no load is in progress!");
           }
         }
       }
-- 
2.20.1