X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2Fchat%2Fbackend%2Fpersistence%2Fkafka%2FKafkaChatHomeService.java;h=f105902ca2a5a651019c685dd63adde015aba71e;hb=3dcdfa896eaecf9802fdc2af4e2bd4951b0f5f6a;hp=556a22683b0d4459c70117e638abf047538578cc;hpb=60aaf90c0288be9af3e73e024970d1f874bf2448;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 556a2268..f105902c 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 @@ -3,80 +3,108 @@ package de.juplo.kafka.chat.backend.persistence.kafka; import de.juplo.kafka.chat.backend.domain.ChatHomeService; import de.juplo.kafka.chat.backend.domain.ChatRoom; import de.juplo.kafka.chat.backend.domain.Message; -import lombok.RequiredArgsConstructor; +import de.juplo.kafka.chat.backend.domain.ShardNotOwnedException; +import de.juplo.kafka.chat.backend.persistence.KafkaLikeShardingStrategy; import lombok.extern.slf4j.Slf4j; import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; +import org.apache.kafka.clients.consumer.ConsumerRecord; +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 reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +import java.time.*; import java.util.*; +import java.util.concurrent.ExecutorService; @Slf4j -public class KafkaChatHomeService implements ChatHomeService, ConsumerRebalanceListener +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 long[] offsets; - private final MessageHandler[] handlers; - private final Map[] chatrooms; + private final ZoneId zoneId; + // private final long[] offsets; Erst mal immer alles neu einlesen + private final boolean[] isShardOwned; + private final Map[] chatRoomMaps; + private final KafkaLikeShardingStrategy shardingStrategy; + + private boolean running; + private volatile boolean loadInProgress; public KafkaChatHomeService( + ExecutorService executorService, Consumer consumer, + Producer producer, String topic, + ZoneId zoneId, int numShards) { log.debug("Creating KafkaChatHomeService"); + this.executorService = executorService; this.consumer = consumer; + this.producer = producer; this.topic = topic; - this.offsets = new long[numShards]; - this.handlers = new MessageHandler[numShards]; - for (int i=0; i< numShards; i++) - { - this.offsets[i] = 0l; - this.handlers[i] = new MessageHandler(new TopicPartition(topic, i)); - } - this.chatrooms = new Map[numShards]; + this.zoneId = zoneId; + // this.offsets = new long[numShards]; + // for (int i=0; i< numShards; i++) + // { + // this.offsets[i] = 0l; + // } + this.isShardOwned = new boolean[numShards]; + this.chatRoomMaps = new Map[numShards]; + this.shardingStrategy = new KafkaLikeShardingStrategy(numShards); } @Override public void onPartitionsAssigned(Collection partitions) { - consumer.endOffsets(partitions).forEach((tp, currentOffset) -> + loadInProgress = true; + + consumer.endOffsets(partitions).forEach((topicPartition, currentOffset) -> { - 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]; + int partition = topicPartition.partition(); + long unseenOffset = 0; // offsets[partition]; - log.info("Reading partition {} from {} -> {}", partition, unseenOffset, currentOffset); - handlers[partition] = new ChatRoomLoadingMessageHandlingStrategy(tp, currentOffset, unseenOffset); + log.info( + "Loading messages from partition {}: start-offset={} -> current-offset={}", + partition, + unseenOffset, + currentOffset); + + // TODO: reuse! Nicht immer alles neu laden, sondern erst ab offsets[partition]! + consumer.seek(topicPartition, unseenOffset); }); + + consumer.resume(partitions); } @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); } @@ -84,117 +112,119 @@ public class KafkaChatHomeService implements ChatHomeService, ConsumerRebalanceL @Override public void onPartitionsLost(Collection partitions) { - log.info("Revoked partitions: {}", partitions); + // TODO: Muss auf den Verlust anders reagiert werden? + onPartitionsRevoked(partitions); } - private void foo() + @Override + public void run() { - 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++) + consumer.subscribe(List.of(topic)); + + running = true; + + try { - chatrooms[shard] = owned.contains(shard) - ? new HashMap<>() - : null; - } - chatroomFlux - .filter(chatRoom -> + while (running) { - if (owned.contains(chatRoom.getShard())) + ConsumerRecords records = consumer.poll(Duration.ofMinutes(5)); + log.info("Fetched {} messages", records.count()); + + if (loadInProgress) { - return true; + for (ConsumerRecord record : records) + { + UUID chatRoomId = UUID.fromString(record.key()); + MessageTo messageTo = record.value(); + ChatRoom chatRoom = chatRoomMaps[record.partition()].get(chatRoomId); + KafkaChatRoomService kafkaChatRoomService = + (KafkaChatRoomService) chatRoom.getChatRoomService(); + 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()); + kafkaChatRoomService.persistMessage(message); + } } else { - log.info("Ignoring not owned chat-room {}", chatRoom); - return false; + if (!records.isEmpty()) + { + throw new IllegalStateException("All owned partions should be paused, when no load is in progress!"); + } } - }) - .toStream() - .forEach(chatroom -> chatrooms[chatroom.getShard()].put(chatroom.getId(), chatroom)); + } + } } - @Override - public Mono putChatRoom(ChatRoom chatRoom) + Mono sendMessage( + UUID chatRoomId, + Message.MessageKey key, + LocalDateTime timestamp, + String text) { - chatrooms[chatRoom.getShard()].put(chatRoom.getId(), chatRoom); - return Mono.just(chatRoom); + int shard = this.shardingStrategy.selectShard(chatRoomId); + TopicPartition tp = new TopicPartition(topic, shard); + ZonedDateTime zdt = ZonedDateTime.of(timestamp, zoneId); + return Mono.create(sink -> + { + ProducerRecord record = + new ProducerRecord<>( + tp.topic(), + tp.partition(), + zdt.toEpochSecond(), + chatRoomId.toString(), + MessageTo.of(key.getUsername(), key.getMessageId(), text)); + + producer.send(record, ((metadata, exception) -> + { + if (metadata != null) + { + // On successful send + Message message = new Message(key, metadata.offset(), timestamp, text); + log.info("Successfully send message {}", message); + sink.success(message); + } + else + { + // On send-failure + log.error( + "Could not send message for chat-room={}, key={}, timestamp={}, text={}: {}", + chatRoomId, + key, + timestamp, + text, + exception); + sink.error(exception); + } + })); + }); } - @Override - public Mono getChatRoom(int shard, UUID id) - { - return Mono.justOrEmpty(chatrooms[shard].get(id)); - } @Override - public Flux getChatRooms(int shard) - { - return Flux.fromStream(chatrooms[shard].values().stream()); - } - - - class MessageHandler - { - - } - - interface MessageHandlingStrategy - { - MessageHandlingStrategy handleMessage(Message message); - } - - - @RequiredArgsConstructor - class NoOpMessageHandlingStrategy implements MessageHandlingStrategy - { - private final TopicPartition tp; - - @Override - public MessageHandlingStrategy handleMessage(Message message) - { - log.warn("Not handling message {} for partition {}", message, tp); - return this; - } - } - - class ChatRoomLoadingMessageHandlingStrategy implements MessageHandlingStrategy + public Mono getChatRoom(int shard, UUID id) { - private final TopicPartition tp; - private final long currentOffset; - private final long unseenOffset; - - ChatRoomLoadingMessageHandlingStrategy(TopicPartition tp, long currentOffset, long unseenOffset) + if (loadInProgress) { - this.tp = tp; - this.currentOffset = currentOffset; - this.unseenOffset = unseenOffset; - - consumer.seek(tp, unseenOffset); + throw new ShardNotOwnedException(shard); } - - @Override - public MessageHandlingStrategy handleMessage(Message message) + else { - // todo - return this; + return Mono.justOrEmpty(chatRoomMaps[shard].get(id)); } } - @RequiredArgsConstructor - class DefaultMessageHandlingStrategy implements MessageHandlingStrategy + @Override + public Flux getChatRooms(int shard) { - private final TopicPartition tp; - - @Override - public MessageHandlingStrategy handleMessage(Message message) + if (loadInProgress) + { + throw new ShardNotOwnedException(shard); + } + else { - chatrooms[tp.partition()].put() - return this; + return Flux.fromStream(chatRoomMaps[shard].values().stream()); } } }