From 5c338c58065988f7841c4ab9ee1b193e754da9b9 Mon Sep 17 00:00:00 2001
From: Kai Moritz <kai@juplo.de>
Date: Thu, 20 Apr 2023 08:50:19 +0200
Subject: [PATCH] NEU
---
.../persistence/kafka/ChatMessageChannel.java | 18 ++++---
.../persistence/kafka/ChatRoomChannel.java | 16 ++++--
.../kafka/KafkaServicesConfiguration.java | 50 +++++++++++++++++++
3 files changed, 75 insertions(+), 9 deletions(-)
diff --git a/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/ChatMessageChannel.java b/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/ChatMessageChannel.java
index 69947a9d..7b19bb6b 100644
--- a/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/ChatMessageChannel.java
+++ b/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/ChatMessageChannel.java
@@ -17,15 +17,13 @@ import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.time.*;
-import java.util.Collection;
-import java.util.List;
-import java.util.Map;
-import java.util.UUID;
+import java.util.*;
+import java.util.concurrent.Callable;
import java.util.stream.IntStream;
@Slf4j
-public class ChatMessageChannel implements Runnable, ConsumerRebalanceListener
+public class ChatMessageChannel implements Callable<Optional<Exception>>, ConsumerRebalanceListener
{
private final String topic;
private final Producer<String, MessageTo> producer;
@@ -155,7 +153,7 @@ public class ChatMessageChannel implements Runnable, ConsumerRebalanceListener
}
@Override
- public void run()
+ public Optional<Exception> call()
{
consumer.subscribe(List.of(topic));
@@ -193,7 +191,15 @@ public class ChatMessageChannel implements Runnable, ConsumerRebalanceListener
log.info("Received WakeupException, exiting!");
running = false;
}
+ catch (Exception e)
+ {
+ log.error("Exiting abnormally!");
+ return Optional.of(e);
+ }
}
+
+ log.info("Exiting normally");
+ return Optional.empty();
}
void loadMessages(ConsumerRecords<String, MessageTo> records)
diff --git a/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/ChatRoomChannel.java b/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/ChatRoomChannel.java
index f9568e70..9ea23b13 100644
--- a/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/ChatRoomChannel.java
+++ b/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/ChatRoomChannel.java
@@ -9,18 +9,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.Mono;
import java.time.*;
import java.util.List;
+import java.util.Optional;
import java.util.UUID;
+import java.util.concurrent.Callable;
@RequiredArgsConstructor
@Slf4j
-public class ChatRoomChannel implements Runnable
+public class ChatRoomChannel implements Callable<Optional<Exception>>
{
private final String topic;
private final Producer<Integer, ChatRoomTo> producer;
@@ -69,7 +70,7 @@ public class ChatRoomChannel implements Runnable
}
@Override
- public void run()
+ public Optional<Exception> call()
{
consumer.assign(List.of(new TopicPartition(topic, 0)));
@@ -92,9 +93,18 @@ public class ChatRoomChannel implements Runnable
log.info("Received WakeupException, exiting!");
running = false;
}
+ catch (Exception e)
+ {
+ log.error("Exiting abnormally!");
+ return Optional.of(e);
+ }
}
+
+ log.info("Exiting normally");
+ return Optional.empty();
}
+
void createChatRoom(ChatRoomInfo chatRoomInfo)
{
UUID id = chatRoomInfo.getId();
diff --git a/src/test/java/de/juplo/kafka/chat/backend/persistence/kafka/KafkaServicesConfiguration.java b/src/test/java/de/juplo/kafka/chat/backend/persistence/kafka/KafkaServicesConfiguration.java
index 55aa6f84..3148c6f0 100644
--- a/src/test/java/de/juplo/kafka/chat/backend/persistence/kafka/KafkaServicesConfiguration.java
+++ b/src/test/java/de/juplo/kafka/chat/backend/persistence/kafka/KafkaServicesConfiguration.java
@@ -9,15 +9,23 @@ import de.juplo.kafka.chat.backend.domain.SimpleChatHome;
import de.juplo.kafka.chat.backend.persistence.KafkaLikeShardingStrategy;
import de.juplo.kafka.chat.backend.persistence.inmemory.InMemoryChatHomeService;
import de.juplo.kafka.chat.backend.persistence.inmemory.InMemoryChatRoomFactory;
+import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.producer.Producer;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+import org.springframework.util.concurrent.ListenableFuture;
import java.time.Clock;
import java.time.ZoneId;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
@ConditionalOnProperty(
@@ -25,8 +33,50 @@ import java.time.ZoneId;
name = "services",
havingValue = "kafka")
@Configuration
+@Slf4j
public class KafkaServicesConfiguration implements ApplicationRunner
{
+ @Autowired
+ ThreadPoolTaskExecutor taskExecutor;
+ @Autowired
+ ConfigurableApplicationContext context;
+
+ @Autowired
+ ChatMessageChannel chatMessageChannel;
+
+ CompletableFuture<Optional<Exception>> chatRoomChannelConsumerJob;
+ CompletableFuture<Optional<Exception>> chatMessageChannelConsumerJob;
+
+
+ @Override
+ public void run(ApplicationArguments args) throws Exception
+ {
+ log.info("Starting the consumer for the ChatRoomChannel");
+ chatRoomChannelConsumerJob = taskExecutor.submitCompletable(chatMessageChannel);
+ chatRoomChannelConsumerJob.thenAccept(exceptionOptional ->
+ {
+ exceptionOptional.ifPresent();
+ log.info("SimpleConsumer exited normally, exit-status: {}", exitStatus);
+ SpringApplication.exit(context, () -> exitStatus);
+ },
+ t ->
+ {
+ log.error("SimpleConsumer exited abnormally!", t);
+ SpringApplication.exit(context, () -> 2);
+ });
+ }
+
+ @PreDestroy
+ public void shutdown() throws ExecutionException, InterruptedException
+ {
+ log.info("Signaling SimpleConsumer to quit its work");
+ kafkaConsumer.wakeup();
+ log.info("Waiting for SimpleConsumer to finish its work");
+ consumerJob.get();
+ log.info("SimpleConsumer finished its work");
+ }
+
+
@Bean
ChatHome kafkaChatHome(
ShardingStrategy shardingStrategy,
--
2.20.1