From f7475320b20be8ba198ba914958e9e4dddf62e11 Mon Sep 17 00:00:00 2001 From: Kai Moritz 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 kafkaChatRoomServiceMap = new HashMap<>(); @@ -33,10 +33,19 @@ class ChatHomeLoader */ boolean handleMessage(ConsumerRecord 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 consumer; private final String topic; // private final long[] offsets; Erst mal immer alles neu einlesen - private final Map[] chatrooms; + private final Map[] kafkaChatRoomServiceMaps; + private final Map[] 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 putChatRoom(ChatRoom chatRoom) { - chatrooms[chatRoom.getShard()].put(chatRoom.getId(), chatRoom); + chatRoomMaps[chatRoom.getShard()].put(chatRoom.getId(), chatRoom); return Mono.just(chatRoom); } @Override public Mono getChatRoom(int shard, UUID id) { - return Mono.justOrEmpty(chatrooms[shard].get(id)); + return Mono.justOrEmpty(chatRoomMaps[shard].get(id)); } @Override public Flux getChatRooms(int shard) { - return Flux.fromStream(chatrooms[shard].values().stream()); + return Flux.fromStream(chatRoomMaps[shard].values().stream()); } } -- 2.20.1