X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2Fchat%2Fbackend%2Fpersistence%2Fkafka%2FChatMessageChannel.java;h=230f8226daffd832b81ddd03b013bcff8287aa12;hb=716ffa6a7665496e614ce6a1671c8e49c562a4c2;hp=7620461785e0e0ed132971bd1cce3b8c6f363c80;hpb=8dc6050f497c742e19cf5363214cf452f31f0017;p=demos%2Fkafka%2Fchat

diff --git a/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/ChatMessageChannel.java b/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/ChatMessageChannel.java
index 76204617..230f8226 100644
--- a/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/ChatMessageChannel.java
+++ b/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/ChatMessageChannel.java
@@ -249,7 +249,37 @@ public class ChatMessageChannel implements Runnable, ConsumerRebalanceListener
 
   void putChatRoom(ChatRoom chatRoom)
   {
-    chatrooms[chatRoom.getShard()].put(chatRoom.getId(), chatRoom);
+    Integer partition = chatRoom.getShard();
+    UUID chatRoomId = chatRoom.getId();
+    ChatRoom existingChatRoom = chatrooms[partition].get(chatRoomId);
+    if (existingChatRoom == null)
+    {
+      log.info(
+          "Creating new chat-room in partition {}: {}",
+          partition,
+          chatRoom);
+      chatrooms[partition].put(chatRoomId, chatRoom);
+    }
+    else
+    {
+      if (chatRoom.getShard() != existingChatRoom.getShard())
+      {
+        throw new IllegalArgumentException(
+            "Could not change the shard of existing chat-room " +
+            chatRoomId + " from " +
+            existingChatRoom.getShard() + " to " +
+            chatRoom.getShard());
+      }
+      else
+      {
+        log.info(
+            "Updating chat-room in partition {}: {} -> {}",
+            partition,
+            existingChatRoom,
+            chatRoom);
+        existingChatRoom.s
+      }
+    }
   }
 
   Mono<ChatRoom> getChatRoom(int shard, UUID id)