From b6b902bc1f64ecd2416c04062d238d550b39c35c Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Thu, 14 Sep 2023 23:17:48 +0200 Subject: [PATCH] refactor: `KafkaChatHomeServiceTest` reuses regular startup-logic --- .../kafka/KafkaServicesApplicationRunner.java | 15 +++--- .../kafka/KafkaServicesConfiguration.java | 31 +++++++++++- .../kafka/KafkaChatHomeServiceTest.java | 48 ++++++++----------- 3 files changed, 58 insertions(+), 36 deletions(-) diff --git a/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/KafkaServicesApplicationRunner.java b/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/KafkaServicesApplicationRunner.java index 8b701807..b635dfc1 100644 --- a/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/KafkaServicesApplicationRunner.java +++ b/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/KafkaServicesApplicationRunner.java @@ -1,17 +1,14 @@ package de.juplo.kafka.chat.backend.implementation.kafka; -import de.juplo.kafka.chat.backend.ChatBackendProperties; import de.juplo.kafka.chat.backend.implementation.kafka.messages.AbstractMessageTo; import jakarta.annotation.PreDestroy; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.apache.kafka.clients.consumer.Consumer; -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.scheduling.concurrent.ThreadPoolTaskExecutor; -import org.springframework.stereotype.Component; import java.util.List; import java.util.concurrent.CompletableFuture; @@ -21,15 +18,14 @@ import java.util.concurrent.CompletableFuture; prefix = "chat.backend", name = "services", havingValue = "kafka") -@Component @RequiredArgsConstructor @Slf4j public class KafkaServicesApplicationRunner implements ApplicationRunner { - private final ChatBackendProperties properties; private final ThreadPoolTaskExecutor taskExecutor; private final ChatRoomChannel chatRoomChannel; private final Consumer chatRoomChannelConsumer; + private final WorkAssignor workAssignor; CompletableFuture chatRoomChannelConsumerJob; @@ -37,8 +33,7 @@ public class KafkaServicesApplicationRunner implements ApplicationRunner @Override public void run(ApplicationArguments args) throws Exception { - List topics = List.of(properties.getKafka().getChatRoomChannelTopic()); - chatRoomChannelConsumer.subscribe(topics, chatRoomChannel); + workAssignor.assignWork(chatRoomChannelConsumer); log.info("Starting the consumer for the ChatRoomChannel"); chatRoomChannelConsumerJob = taskExecutor .submitCompletable(chatRoomChannel) @@ -58,4 +53,10 @@ public class KafkaServicesApplicationRunner implements ApplicationRunner chatRoomChannelConsumerJob.join(); log.info("Joined the consumer of the ChatRoomChannel"); } + + + interface WorkAssignor + { + void assignWork(Consumer consumer); + } } diff --git a/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/KafkaServicesConfiguration.java b/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/KafkaServicesConfiguration.java index d63111a7..d17c79c9 100644 --- a/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/KafkaServicesConfiguration.java +++ b/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/KafkaServicesConfiguration.java @@ -18,10 +18,12 @@ import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.kafka.support.serializer.JsonDeserializer; import org.springframework.kafka.support.serializer.JsonSerializer; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import java.time.Clock; import java.time.ZoneId; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.Properties; @@ -34,7 +36,34 @@ import java.util.Properties; public class KafkaServicesConfiguration { @Bean - ChatHomeService kafkaChatHome( + KafkaServicesApplicationRunner kafkaServicesApplicationRunner( + ThreadPoolTaskExecutor taskExecutor, + ChatRoomChannel chatRoomChannel, + Consumer chatRoomChannelConsumer, + KafkaServicesApplicationRunner.WorkAssignor workAssignor) + { + return new KafkaServicesApplicationRunner( + taskExecutor, + chatRoomChannel, + chatRoomChannelConsumer, + workAssignor); + } + + @Bean + KafkaServicesApplicationRunner.WorkAssignor workAssignor( + ChatBackendProperties properties, + ChatRoomChannel chatRoomChannel) + { + return consumer -> + { + List topics = + List.of(properties.getKafka().getChatRoomChannelTopic()); + consumer.subscribe(topics, chatRoomChannel); + }; + } + + @Bean + ChatHomeService kafkaChatHome( ChatBackendProperties properties, ChatRoomChannel chatRoomChannel) { diff --git a/src/test/java/de/juplo/kafka/chat/backend/implementation/kafka/KafkaChatHomeServiceTest.java b/src/test/java/de/juplo/kafka/chat/backend/implementation/kafka/KafkaChatHomeServiceTest.java index c120e465..204b34e6 100644 --- a/src/test/java/de/juplo/kafka/chat/backend/implementation/kafka/KafkaChatHomeServiceTest.java +++ b/src/test/java/de/juplo/kafka/chat/backend/implementation/kafka/KafkaChatHomeServiceTest.java @@ -3,26 +3,25 @@ package de.juplo.kafka.chat.backend.implementation.kafka; import de.juplo.kafka.chat.backend.ChatBackendProperties; import de.juplo.kafka.chat.backend.domain.ChatHomeServiceWithShardsTest; import lombok.extern.slf4j.Slf4j; -import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.TopicPartition; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.BeforeAll; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.DefaultApplicationArguments; import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration; import org.springframework.boot.autoconfigure.task.TaskExecutionAutoConfiguration; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.boot.test.context.TestConfiguration; import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Import; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.support.SendResult; import org.springframework.kafka.test.context.EmbeddedKafka; -import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import java.time.Clock; import java.util.List; -import java.util.concurrent.CompletableFuture; import static de.juplo.kafka.chat.backend.domain.ChatHomeServiceWithShardsTest.NUM_SHARDS; import static de.juplo.kafka.chat.backend.implementation.kafka.KafkaChatHomeServiceTest.TOPIC; @@ -31,11 +30,11 @@ import static de.juplo.kafka.chat.backend.implementation.kafka.KafkaChatHomeServ @SpringBootTest( classes = { KafkaChatHomeServiceTest.KafkaChatHomeTestConfiguration.class, - KafkaServicesConfiguration.class, KafkaAutoConfiguration.class, TaskExecutionAutoConfiguration.class, }, properties = { + "spring.main.allow-bean-definition-overriding=true", "chat.backend.services=kafka", "chat.backend.kafka.client-id-PREFIX=TEST", "chat.backend.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}", @@ -49,13 +48,25 @@ public class KafkaChatHomeServiceTest extends ChatHomeServiceWithShardsTest { final static String TOPIC = "KAFKA_CHAT_HOME_TEST"; - static CompletableFuture CONSUMER_JOB; - @TestConfiguration @EnableConfigurationProperties(ChatBackendProperties.class) + @Import(KafkaServicesConfiguration.class) static class KafkaChatHomeTestConfiguration { + @Bean + KafkaServicesApplicationRunner.WorkAssignor workAssignor( + ChatRoomChannel chatRoomChannel) + { + return consumer -> + { + List assignedPartitions = + List.of(new TopicPartition(TOPIC, 2)); + consumer.assign(assignedPartitions); + chatRoomChannel.onPartitionsAssigned(assignedPartitions); + }; + } + @Bean Clock clock() { @@ -65,28 +76,13 @@ public class KafkaChatHomeServiceTest extends ChatHomeServiceWithShardsTest @BeforeAll - public static void sendAndLoadStoredData( - @Autowired KafkaTemplate messageTemplate, - @Autowired Consumer chatRoomChannelConsumer, - @Autowired ThreadPoolTaskExecutor taskExecutor, - @Autowired ChatRoomChannel chatRoomChannel) + public static void sendAndLoadStoredData(@Autowired KafkaTemplate messageTemplate) { send(messageTemplate, "5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\": \"5c73531c-6fc4-426c-adcb-afc5c140a0f7\", \"shard\": 2, \"name\": \"FOO\" }", "command_create_chatroom"); send(messageTemplate,"5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 1, \"user\" : \"peter\", \"text\" : \"Hallo, ich heiße Peter!\" }", "event_chatmessage_received"); send(messageTemplate,"5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 1, \"user\" : \"ute\", \"text\" : \"Ich bin Ute...\" }", "event_chatmessage_received"); send(messageTemplate,"5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 2, \"user\" : \"peter\", \"text\" : \"Willst du mit mir gehen?\" }", "event_chatmessage_received"); send(messageTemplate,"5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 1, \"user\" : \"klaus\", \"text\" : \"Ja? Nein? Vielleicht??\" }", "event_chatmessage_received"); - - List assignedPartitions = List.of(new TopicPartition(TOPIC, 2)); - chatRoomChannelConsumer.assign(assignedPartitions); - chatRoomChannel.onPartitionsAssigned(assignedPartitions); - CONSUMER_JOB = taskExecutor - .submitCompletable(chatRoomChannel) - .exceptionally(e -> - { - log.error("The consumer for the ChatRoomChannel exited abnormally!", e); - return null; - }); } static void send(KafkaTemplate kafkaTemplate, String key, String value, String typeId) @@ -102,12 +98,8 @@ public class KafkaChatHomeServiceTest extends ChatHomeServiceWithShardsTest } @AfterAll - static void joinConsumerJob(@Autowired Consumer chatRoomChannelConsumer) + static void joinConsumerJob(@Autowired KafkaServicesApplicationRunner applicationRunner) { - log.info("Signaling the consumer of the CahtRoomChannel to quit its work"); - chatRoomChannelConsumer.wakeup(); - log.info("Waiting for the consumer of the ChatRoomChannel to finish its work"); - CONSUMER_JOB.join(); - log.info("Joined the consumer of the ChatRoomChannel"); + applicationRunner.joinChatRoomChannelConsumerJob(); } } -- 2.20.1