X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2Fchat%2Fbackend%2Fpersistence%2Fkafka%2FKafkaChatHomeService.java;h=e23f08db4677975bac0f5fcbf286a3278ec5d099;hb=9aaca321143c4fd08859dc14cdd7e602cc1714f6;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..e23f08db 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,28 +13,31 @@ 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 consumer; - private final Producer producer; - private final String topic; + private final String chatRoomsTopic; + private final Consumer chatRoomsConsumer; + private final Producer chatRoomsProducer; + private final String chatMessagesTopic; + private final Consumer chatMessagesConsumer; + private final Producer chatMessagesProducer; private final ZoneId zoneId; private final int numShards; private final boolean[] isShardOwned; private final long[] currentOffset; private final long[] nextOffset; - private final Map[] chatRoomMaps; + private final Map[] chatrooms; private final KafkaLikeShardingStrategy shardingStrategy; private boolean running; @@ -42,24 +45,28 @@ public class KafkaChatHomeService implements ChatHomeService, Runnable, Consumer public KafkaChatHomeService( - ExecutorService executorService, - Consumer consumer, - Producer producer, - String topic, + String chatRoomsTopic, + Consumer chatRoomsConsumer, + Producer chatRoomsProducer, + String chatMessagesTopic, + Consumer chatMessagesConsumer, + Producer chatMessagesProducer, ZoneId zoneId, int numShards) { log.debug("Creating KafkaChatHomeService"); - this.executorService = executorService; - this.consumer = consumer; - this.producer = producer; - this.topic = topic; + this.chatRoomsTopic = chatRoomsTopic; + this.chatRoomsConsumer = chatRoomsConsumer; + this.chatRoomsProducer = chatRoomsProducer; + this.chatMessagesTopic = chatMessagesTopic; + this.chatMessagesConsumer = chatMessagesConsumer; + this.chatMessagesProducer = chatMessagesProducer; this.zoneId = zoneId; this.numShards = numShards; 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,9 +74,10 @@ public class KafkaChatHomeService implements ChatHomeService, Runnable, Consumer @Override public void onPartitionsAssigned(Collection partitions) { + log.info("Newly assigned partitions! Pausing normal operations..."); loadInProgress = true; - consumer.endOffsets(partitions).forEach((topicPartition, currentOffset) -> + chatMessagesConsumer.endOffsets(partitions).forEach((topicPartition, currentOffset) -> { int partition = topicPartition.partition(); isShardOwned[partition] = true; @@ -81,10 +89,10 @@ public class KafkaChatHomeService implements ChatHomeService, Runnable, Consumer nextOffset[partition], currentOffset); - consumer.seek(topicPartition, nextOffset[partition]); + chatMessagesConsumer.seek(topicPartition, nextOffset[partition]); }); - consumer.resume(partitions); + chatMessagesConsumer.resume(partitions); } @Override @@ -109,54 +117,26 @@ public class KafkaChatHomeService implements ChatHomeService, Runnable, Consumer @Override public void run() { - consumer.subscribe(List.of(topic)); + chatMessagesConsumer.subscribe(List.of(chatMessagesTopic)); running = true; - try + while (running) { - while (running) + try { - ConsumerRecords records = consumer.poll(Duration.ofMinutes(5)); + ConsumerRecords records = chatMessagesConsumer.poll(Duration.ofMinutes(5)); log.info("Fetched {} messages", records.count()); if (loadInProgress) { - for (ConsumerRecord 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 +148,59 @@ public class KafkaChatHomeService implements ChatHomeService, Runnable, Consumer } } } + catch (WakeupException e) + { + } + catch (RecordDeserializationException e) + { + } } } + void loadMessages(ConsumerRecords records) + { + for (ConsumerRecord 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() + { + chatMessagesConsumer.pause(IntStream + .range(0, numShards) + .filter(shard -> isShardOwned[shard]) + .mapToObj(shard -> new TopicPartition(chatMessagesTopic, shard)) + .toList()); + } + Mono sendMessage( UUID chatRoomId, Message.MessageKey key, @@ -178,7 +208,7 @@ public class KafkaChatHomeService implements ChatHomeService, Runnable, Consumer String text) { int shard = this.shardingStrategy.selectShard(chatRoomId); - TopicPartition tp = new TopicPartition(topic, shard); + TopicPartition tp = new TopicPartition(chatMessagesTopic, shard); ZonedDateTime zdt = ZonedDateTime.of(timestamp, zoneId); return Mono.create(sink -> { @@ -190,7 +220,7 @@ public class KafkaChatHomeService implements ChatHomeService, Runnable, Consumer chatRoomId.toString(), MessageTo.of(key.getUsername(), key.getMessageId(), text)); - producer.send(record, ((metadata, exception) -> + chatMessagesProducer.send(record, ((metadata, exception) -> { if (metadata != null) { @@ -216,6 +246,14 @@ public class KafkaChatHomeService implements ChatHomeService, Runnable, Consumer } + public void putChatRoom(ChatRoom chatRoom) + { + + ProducerRecord record = new ProducerRecord<>(chatRoom.getShard(), ); + // TODO: Nachricht senden! + chatrooms[chatRoom.getShard()].put(chatRoom.getId(), chatRoom); + } + @Override public Mono getChatRoom(int shard, UUID id) { @@ -225,7 +263,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 +276,7 @@ public class KafkaChatHomeService implements ChatHomeService, Runnable, Consumer } else { - return Flux.fromStream(chatRoomMaps[shard].values().stream()); + return Flux.fromStream(chatrooms[shard].values().stream()); } } }