X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2Fchat%2Fbackend%2Fpersistence%2Fkafka%2FKafkaChatHomeService.java;h=22cd74b75ac4c879fbbd0a0182491a3e87414655;hb=915ed8f85459da3c95f86b6351a3d7129668bc8e;hp=b4dccbd1b2538c2e4dc2fcfa3aed1ad25834eecf;hpb=3ff99ab559bf06b97c4c282a7933df951cbeaa8f;p=demos%2Fkafka%2Fchat

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 b4dccbd1..22cd74b7 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
@@ -13,19 +13,19 @@ import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.RecordDeserializationException;
+import org.apache.kafka.common.errors.WakeupException;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import java.time.*;
 import java.util.*;
-import java.util.concurrent.ExecutorService;
 import java.util.stream.IntStream;
 
 
 @Slf4j
 public class KafkaChatHomeService implements ChatHomeService, Runnable, ConsumerRebalanceListener
 {
-  private final ExecutorService executorService;
   private final Consumer<String, MessageTo> consumer;
   private final Producer<String, MessageTo> producer;
   private final String topic;
@@ -34,7 +34,7 @@ public class KafkaChatHomeService implements ChatHomeService, Runnable, Consumer
   private final boolean[] isShardOwned;
   private final long[] currentOffset;
   private final long[] nextOffset;
-  private final Map<UUID, ChatRoom>[] chatRoomMaps;
+  private final Map<UUID, ChatRoom>[] chatrooms;
   private final KafkaLikeShardingStrategy shardingStrategy;
 
   private boolean running;
@@ -42,7 +42,6 @@ public class KafkaChatHomeService implements ChatHomeService, Runnable, Consumer
 
 
   public KafkaChatHomeService(
-    ExecutorService executorService,
     Consumer<String, MessageTo> consumer,
     Producer<String, MessageTo> producer,
     String topic,
@@ -50,7 +49,6 @@ public class KafkaChatHomeService implements ChatHomeService, Runnable, Consumer
     int numShards)
   {
     log.debug("Creating KafkaChatHomeService");
-    this.executorService = executorService;
     this.consumer = consumer;
     this.producer = producer;
     this.topic = topic;
@@ -59,7 +57,7 @@ public class KafkaChatHomeService implements ChatHomeService, Runnable, Consumer
     this.isShardOwned = new boolean[numShards];
     this.currentOffset = new long[numShards];
     this.nextOffset = new long[numShards];
-    this.chatRoomMaps = new Map[numShards];
+    this.chatrooms = new Map[numShards];
     this.shardingStrategy = new KafkaLikeShardingStrategy(numShards);
   }
 
@@ -67,6 +65,7 @@ public class KafkaChatHomeService implements ChatHomeService, Runnable, Consumer
   @Override
   public void onPartitionsAssigned(Collection<TopicPartition> partitions)
   {
+    log.info("Newly assigned partitions! Pausing normal operations...");
     loadInProgress = true;
 
     consumer.endOffsets(partitions).forEach((topicPartition, currentOffset) ->
@@ -113,50 +112,22 @@ public class KafkaChatHomeService implements ChatHomeService, Runnable, Consumer
 
     running = true;
 
-    try
+    while (running)
     {
-      while (running)
+      try
       {
         ConsumerRecords<String, MessageTo> records = consumer.poll(Duration.ofMinutes(5));
         log.info("Fetched {} messages", records.count());
 
         if (loadInProgress)
         {
-          for (ConsumerRecord<String, MessageTo> record : records)
-          {
-            nextOffset[record.partition()] = record.offset() + 1;
-            UUID chatRoomId = UUID.fromString(record.key());
-            MessageTo messageTo = record.value();
-
-            Message.MessageKey key = Message.MessageKey.of(messageTo.getUser(), messageTo.getId());
-
-            Instant instant = Instant.ofEpochSecond(record.timestamp());
-            LocalDateTime timestamp = LocalDateTime.ofInstant(instant, zoneId);
-
-            Message message = new Message(key, record.offset(), timestamp, messageTo.getText());
+          loadMessages(records);
 
-            ChatRoom chatRoom = chatRoomMaps[record.partition()].get(chatRoomId);
-            KafkaChatRoomService kafkaChatRoomService =
-                (KafkaChatRoomService) chatRoom.getChatRoomService();
-
-            kafkaChatRoomService.persistMessage(message);
-          }
-
-          if (IntStream
-            .range(0, numShards)
-            .filter(shard -> isShardOwned[shard])
-            .mapToObj(shard -> nextOffset[shard] >= currentOffset[shard])
-            .collect(
-                () -> Boolean.TRUE,
-                (acc, v) -> Boolean.valueOf(acc && v),
-                (a, b) -> Boolean.valueOf(a && b)))
+          if (isLoadingCompleted())
           {
             log.info("Loading of messages completed! Pausing all owned partitions...");
-            consumer.pause(IntStream
-                .range(0, numShards)
-                .filter(shard -> isShardOwned[shard])
-                .mapToObj(shard -> new TopicPartition(topic, shard))
-                .toList());
+            pauseAllOwnedPartions();
+            log.info("Resuming normal operations...");
             loadInProgress = false;
           }
         }
@@ -168,9 +139,59 @@ public class KafkaChatHomeService implements ChatHomeService, Runnable, Consumer
           }
         }
       }
+      catch (WakeupException e)
+      {
+      }
+      catch (RecordDeserializationException e)
+      {
+      }
+    }
+  }
+
+  void loadMessages(ConsumerRecords<String, MessageTo> records)
+  {
+    for (ConsumerRecord<String, MessageTo> record : records)
+    {
+      nextOffset[record.partition()] = record.offset() + 1;
+      UUID chatRoomId = UUID.fromString(record.key());
+      MessageTo messageTo = record.value();
+
+      Message.MessageKey key = Message.MessageKey.of(messageTo.getUser(), messageTo.getId());
+
+      Instant instant = Instant.ofEpochSecond(record.timestamp());
+      LocalDateTime timestamp = LocalDateTime.ofInstant(instant, zoneId);
+
+      Message message = new Message(key, record.offset(), timestamp, messageTo.getText());
+
+      ChatRoom chatRoom = chatrooms[record.partition()].get(chatRoomId);
+      KafkaChatRoomService kafkaChatRoomService =
+          (KafkaChatRoomService) chatRoom.getChatRoomService();
+
+      kafkaChatRoomService.persistMessage(message);
     }
   }
 
+  boolean isLoadingCompleted()
+  {
+    return IntStream
+        .range(0, numShards)
+        .filter(shard -> isShardOwned[shard])
+        .mapToObj(shard -> nextOffset[shard] >= currentOffset[shard])
+        .collect(
+            () -> Boolean.TRUE,
+            (acc, v) -> Boolean.valueOf(acc && v),
+            (a, b) -> Boolean.valueOf(a && b));
+  }
+
+  void pauseAllOwnedPartions()
+  {
+    consumer.pause(IntStream
+        .range(0, numShards)
+        .filter(shard -> isShardOwned[shard])
+        .mapToObj(shard -> new TopicPartition(topic, shard))
+        .toList());
+  }
+
   Mono<Message> sendMessage(
       UUID chatRoomId,
       Message.MessageKey key,
@@ -216,6 +237,12 @@ public class KafkaChatHomeService implements ChatHomeService, Runnable, Consumer
   }
 
 
+  public void putChatRoom(ChatRoom chatRoom)
+  {
+    // TODO: Nachricht senden!
+    chatrooms[chatRoom.getShard()].put(chatRoom.getId(), chatRoom);
+  }
+
   @Override
   public Mono<ChatRoom> getChatRoom(int shard, UUID id)
   {
@@ -225,7 +252,7 @@ public class KafkaChatHomeService implements ChatHomeService, Runnable, Consumer
     }
     else
     {
-      return Mono.justOrEmpty(chatRoomMaps[shard].get(id));
+      return Mono.justOrEmpty(chatrooms[shard].get(id));
     }
   }
 
@@ -238,7 +265,7 @@ public class KafkaChatHomeService implements ChatHomeService, Runnable, Consumer
     }
     else
     {
-      return Flux.fromStream(chatRoomMaps[shard].values().stream());
+      return Flux.fromStream(chatrooms[shard].values().stream());
     }
   }
 }