From: Kai Moritz Date: Sun, 20 Aug 2023 10:20:55 +0000 (+0200) Subject: fix: GREEN - Correct handling of unknown chat-rooms in the Kafka-version X-Git-Tag: wip-inmemory-chat-home-service~11 X-Git-Url: https://juplo.de/gitweb/?a=commitdiff_plain;h=5d6fc3eee1a20ea5c8a808320ef12b51b84f54be;p=demos%2Fkafka%2Fchat fix: GREEN - Correct handling of unknown chat-rooms in the Kafka-version --- diff --git a/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/ChatRoomChannel.java b/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/ChatRoomChannel.java index 7f93ad8b..7659d1e1 100644 --- a/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/ChatRoomChannel.java +++ b/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/ChatRoomChannel.java @@ -360,6 +360,14 @@ public class ChatRoomChannel implements Runnable, ConsumerRebalanceListener } } + int[] getOwnedShards() + { + return IntStream + .range(0, numShards) + .filter(shard -> isShardOwned[shard]) + .toArray(); + } + Mono getChatRoom(int shard, UUID id) { if (loadInProgress) diff --git a/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/KafkaChatHome.java b/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/KafkaChatHome.java index 67990409..ab72269d 100644 --- a/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/KafkaChatHome.java +++ b/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/KafkaChatHome.java @@ -2,6 +2,7 @@ package de.juplo.kafka.chat.backend.persistence.kafka; import de.juplo.kafka.chat.backend.domain.ChatHome; import de.juplo.kafka.chat.backend.domain.ChatRoom; +import de.juplo.kafka.chat.backend.domain.UnknownChatroomException; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.apache.kafka.common.utils.Utils; @@ -23,7 +24,9 @@ public class KafkaChatHome implements ChatHome public Mono getChatRoom(UUID id) { int shard = selectShard(id); - return chatRoomChannel.getChatRoom(shard, id); + return chatRoomChannel + .getChatRoom(shard, id) + .switchIfEmpty(Mono.error(() -> new UnknownChatroomException(id))); } int selectShard(UUID chatRoomId)