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=7ebf0493056ae19450cbb3496821cc5a9c8d6a18;hpb=f14af7787fd519cffcdcd83cd375fb7477b4ced2;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 7ebf0493..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 @@ -2,28 +2,43 @@ 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 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 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.ZoneId; +import java.time.*; import java.util.*; +import java.util.stream.IntStream; @Slf4j -public class KafkaChatHomeService implements ChatHomeService, ConsumerRebalanceListener +public class KafkaChatHomeService implements ChatHomeService, Runnable, ConsumerRebalanceListener { private final Consumer consumer; private final Producer producer; private final String topic; private final ZoneId zoneId; - // private final long[] offsets; Erst mal immer alles neu einlesen - private final ChatHomeLoader[] chatHomeLoaders; - private final Map[] chatRoomMaps; + private final int numShards; + private final boolean[] isShardOwned; + private final long[] currentOffset; + private final long[] nextOffset; + private final Map[] chatrooms; + private final KafkaLikeShardingStrategy shardingStrategy; + + private boolean running; + private volatile boolean loadInProgress; public KafkaChatHomeService( @@ -38,117 +53,219 @@ public class KafkaChatHomeService implements ChatHomeService, ConsumerRebalanceL this.producer = producer; this.topic = topic; this.zoneId = zoneId; - // this.offsets = new long[numShards]; - // for (int i=0; i< numShards; i++) - // { - // this.offsets[i] = 0l; - // } - this.chatHomeLoaders = new ChatHomeLoader[numShards]; - this.chatRoomMaps = new Map[numShards]; + this.numShards = numShards; + this.isShardOwned = new boolean[numShards]; + this.currentOffset = new long[numShards]; + this.nextOffset = new long[numShards]; + this.chatrooms = new Map[numShards]; + this.shardingStrategy = new KafkaLikeShardingStrategy(numShards); } @Override public void onPartitionsAssigned(Collection partitions) { + log.info("Newly assigned partitions! Pausing normal operations..."); + loadInProgress = true; + consumer.endOffsets(partitions).forEach((topicPartition, currentOffset) -> { - if (!topicPartition.topic().equals(topic)) - { - log.warn("Ignoring partition from unwanted topic: {}", topicPartition); - return; - } - int partition = topicPartition.partition(); - long unseenOffset = 0; // offsets[partition]; + isShardOwned[partition] = true; + this.currentOffset[partition] = currentOffset; log.info( - "Loading messages from partition {}: start-offset={} -> current-offset={}", + "Partition assigned: {} - loading messages: next={} -> current={}", partition, - unseenOffset, + nextOffset[partition], currentOffset); - // TODO: reuse! Nicht immer alles neu laden, sondern erst ab offsets[partition]! - consumer.seek(topicPartition, unseenOffset); - chatHomeLoaders[partition] = new ChatHomeLoader( - producer, - currentOffset, - zoneId); + consumer.seek(topicPartition, nextOffset[partition]); }); + + consumer.resume(partitions); } @Override public void onPartitionsRevoked(Collection partitions) { - partitions.forEach(tp -> + partitions.forEach(topicPartition -> { - if (!tp.topic().equals(topic)) - { - log.warn("Ignoring partition from unwanted topic: {}", tp); - return; - } - - int partition = tp.partition(); - long unseenOffset = offsets[partition]; - - log.info("Reading partition {} from {} -> {}", partition, unseenOffset, currentOffset); + int partition = topicPartition.partition(); + isShardOwned[partition] = false; + log.info("Partition revoked: {} - next={}", partition, nextOffset[partition]); }); - log.info("Revoked partitions: {}", partitions); } @Override public void onPartitionsLost(Collection partitions) { - log.info("Revoked partitions: {}", partitions); + log.warn("Lost 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; + + while (running) { - chatRoomMaps[shard] = owned.contains(shard) - ? new HashMap<>() - : null; + try + { + ConsumerRecords records = consumer.poll(Duration.ofMinutes(5)); + log.info("Fetched {} messages", records.count()); + + if (loadInProgress) + { + loadMessages(records); + + if (isLoadingCompleted()) + { + log.info("Loading of messages completed! Pausing all owned partitions..."); + pauseAllOwnedPartions(); + log.info("Resuming normal operations..."); + loadInProgress = false; + } + } + else + { + if (!records.isEmpty()) + { + throw new IllegalStateException("All owned partitions should be paused, when no load is in progress!"); + } + } + } + catch (WakeupException e) + { + } + catch (RecordDeserializationException e) + { + } } - chatroomFlux - .filter(chatRoom -> + } + + 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() + { + consumer.pause(IntStream + .range(0, numShards) + .filter(shard -> isShardOwned[shard]) + .mapToObj(shard -> new TopicPartition(topic, shard)) + .toList()); + } + + Mono sendMessage( + UUID chatRoomId, + Message.MessageKey key, + LocalDateTime timestamp, + String text) + { + 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 (owned.contains(chatRoom.getShard())) + if (metadata != null) { - return true; + // On successful send + Message message = new Message(key, metadata.offset(), timestamp, text); + log.info("Successfully send message {}", message); + sink.success(message); } else { - log.info("Ignoring not owned chat-room {}", chatRoom); - return false; + // On send-failure + log.error( + "Could not send message for chat-room={}, key={}, timestamp={}, text={}: {}", + chatRoomId, + key, + timestamp, + text, + exception); + sink.error(exception); } - }) - .toStream() - .forEach(chatroom -> chatRoomMaps[chatroom.getShard()].put(chatroom.getId(), chatroom)); + })); + }); } - @Override - public Mono putChatRoom(ChatRoom chatRoom) + + public void putChatRoom(ChatRoom chatRoom) { - chatRoomMaps[chatRoom.getShard()].put(chatRoom.getId(), chatRoom); - return Mono.just(chatRoom); + // TODO: Nachricht senden! + chatrooms[chatRoom.getShard()].put(chatRoom.getId(), chatRoom); } @Override public Mono getChatRoom(int shard, UUID id) { - return Mono.justOrEmpty(chatRoomMaps[shard].get(id)); + if (loadInProgress) + { + throw new ShardNotOwnedException(shard); + } + else + { + return Mono.justOrEmpty(chatrooms[shard].get(id)); + } } @Override public Flux getChatRooms(int shard) { - return Flux.fromStream(chatRoomMaps[shard].values().stream()); + if (loadInProgress) + { + throw new ShardNotOwnedException(shard); + } + else + { + return Flux.fromStream(chatrooms[shard].values().stream()); + } } }