From faeec2b848a9e4a25fed7ca1e771068ea8c59583 Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Sun, 26 Feb 2023 15:48:25 +0100 Subject: [PATCH] WIP --- .../persistence/kafka/KafkaChatHomeService.java | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 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 2e3b42f6..b2ed2842 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,6 @@ public class KafkaChatHomeService implements ChatHomeService, ConsumerRebalanceL private final Consumer consumer; private final String topic; private final long[] offsets; - private final MessageHandler[] handlers; private final Map[] chatrooms; @@ -31,13 +30,14 @@ public class KafkaChatHomeService implements ChatHomeService, ConsumerRebalanceL this.consumer = consumer; this.topic = topic; this.offsets = new long[numShards]; - this.handlers = new MessageHandler[numShards]; + this.chatrooms = new Map<>[numShards]; for (int i=0; i< numShards; i++) { this.offsets[i] = 0l; - this.handlers[i] = new MessageHandler(consumer, new TopicPartition(topic, i)); + this.chatrooms[i] = chat } this.chatrooms = new Map[numShards]; + this.chatrooms = new Map[numShards]; } @@ -55,7 +55,11 @@ public class KafkaChatHomeService implements ChatHomeService, ConsumerRebalanceL int partition = tp.partition(); long unseenOffset = offsets[partition]; - log.info("Reading partition {} from {} -> {}", partition, unseenOffset, currentOffset); + log.info( + "Loading messages from partition {}: start-offset={} -> current-offset={}", + partition, + unseenOffset, + currentOffset); handlers[partition] = new ChatRoomLoadingMessageHandlingStrategy(tp, currentOffset, unseenOffset); }); } -- 2.20.1