From b44e10be71112a45855d04add6497e40d4c72d24 Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Sun, 26 Feb 2023 19:26:50 +0100 Subject: [PATCH] WIP:compiles! --- ...ChatRoomActiveMessageHandlingStrategy.java | 5 +- .../kafka/KafkaChatHomeService.java | 53 +++---------------- 2 files changed, 9 insertions(+), 49 deletions(-) diff --git a/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/ChatRoomActiveMessageHandlingStrategy.java b/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/ChatRoomActiveMessageHandlingStrategy.java index cfd0e4ec..04564b9c 100644 --- a/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/ChatRoomActiveMessageHandlingStrategy.java +++ b/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/ChatRoomActiveMessageHandlingStrategy.java @@ -1,9 +1,8 @@ package de.juplo.kafka.chat.backend.persistence.kafka; import de.juplo.kafka.chat.backend.domain.Message; -import de.juplo.kafka.chat.backend.domain.MessageMutationException; import lombok.RequiredArgsConstructor; -import lombok.extern.log4j.Log4j; +import lombok.extern.slf4j.Slf4j; import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.TopicPartition; @@ -15,7 +14,7 @@ import java.util.UUID; @RequiredArgsConstructor -@Log4j +@Slf4j class ChatRoomActiveMessageHandlingStrategy implements MessageHandlingStrategy { private final KafkaChatRoomService kafkaChatRoomService; 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 7ebf0493..eadd7629 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 @@ -80,18 +80,16 @@ public class KafkaChatHomeService implements ChatHomeService, ConsumerRebalanceL @Override public void onPartitionsRevoked(Collection partitions) { - partitions.forEach(tp -> + partitions.forEach(topicPartition -> { - if (!tp.topic().equals(topic)) + if (!topicPartition.topic().equals(topic)) { - log.warn("Ignoring partition from unwanted topic: {}", tp); + log.warn("Ignoring partition from unwanted topic: {}", topicPartition); return; } - int partition = tp.partition(); - long unseenOffset = offsets[partition]; - - log.info("Reading partition {} from {} -> {}", partition, unseenOffset, currentOffset); + int partition = topicPartition.partition(); + // long unseenOffset = offsets[partition]; TODO: Offset merken...? }); log.info("Revoked partitions: {}", partitions); } @@ -99,45 +97,8 @@ public class KafkaChatHomeService implements ChatHomeService, ConsumerRebalanceL @Override public void onPartitionsLost(Collection partitions) { - log.info("Revoked partitions: {}", partitions); - } - - private void foo() - { - Set owned = Arrays - .stream(ownedShards) - .collect( - () -> new HashSet<>(), - (set, i) -> set.add(i), - (a, b) -> a.addAll(b)); - for (int shard = 0; shard < numShards; shard++) - { - chatRoomMaps[shard] = owned.contains(shard) - ? new HashMap<>() - : null; - } - chatroomFlux - .filter(chatRoom -> - { - if (owned.contains(chatRoom.getShard())) - { - return true; - } - else - { - log.info("Ignoring not owned chat-room {}", chatRoom); - return false; - } - }) - .toStream() - .forEach(chatroom -> chatRoomMaps[chatroom.getShard()].put(chatroom.getId(), chatroom)); - } - - @Override - public Mono putChatRoom(ChatRoom chatRoom) - { - chatRoomMaps[chatRoom.getShard()].put(chatRoom.getId(), chatRoom); - return Mono.just(chatRoom); + // TODO: Muss auf den Verlust anders reagiert werden? + onPartitionsRevoked(partitions); } @Override -- 2.20.1