fix: `ConsumerTaskRunner` waits until the data-loading is finished
[demos/kafka/chat] / src / test / java / de / juplo / kafka / chat / backend / implementation / kafka / KafkaChatHomeServiceTest.java
1 package de.juplo.kafka.chat.backend.implementation.kafka;
2
3 import de.juplo.kafka.chat.backend.domain.ChatHomeServiceWithShardsTest;
4 import lombok.extern.slf4j.Slf4j;
5 import org.junit.jupiter.api.AfterAll;
6 import org.junit.jupiter.api.BeforeAll;
7 import org.springframework.beans.factory.annotation.Autowired;
8 import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration;
9 import org.springframework.boot.autoconfigure.task.TaskExecutionAutoConfiguration;
10 import org.springframework.boot.test.context.SpringBootTest;
11 import org.springframework.kafka.core.KafkaTemplate;
12 import org.springframework.kafka.test.context.EmbeddedKafka;
13
14 import static de.juplo.kafka.chat.backend.domain.ChatHomeServiceWithShardsTest.NUM_SHARDS;
15 import static de.juplo.kafka.chat.backend.implementation.kafka.KafkaChatHomeServiceTest.DATA_TOPIC;
16 import static de.juplo.kafka.chat.backend.implementation.kafka.KafkaChatHomeServiceTest.INFO_TOPIC;
17
18
19 @SpringBootTest(
20     classes = {
21         KafkaTestUtils.KafkaTestConfiguration.class,
22         KafkaAutoConfiguration.class,
23         TaskExecutionAutoConfiguration.class,
24     },
25     properties = {
26         "spring.main.allow-bean-definition-overriding=true",
27         "chat.backend.services=kafka",
28         "chat.backend.kafka.client-id-PREFIX=TEST",
29         "chat.backend.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
30         "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
31         "chat.backend.kafka.info-channel-topic=" + INFO_TOPIC,
32         "chat.backend.kafka.data-channel-topic=" + DATA_TOPIC,
33         "chat.backend.kafka.num-partitions=" + NUM_SHARDS,
34 })
35 @EmbeddedKafka(
36     topics = { INFO_TOPIC, DATA_TOPIC },
37     partitions = 10)
38 @Slf4j
39 public class KafkaChatHomeServiceTest extends ChatHomeServiceWithShardsTest
40 {
41   final static String INFO_TOPIC = "KAFKA_CHAT_HOME_TEST_INFO";
42   final static String DATA_TOPIC = "KAFKA_CHAT_HOME_TEST_DATA";
43
44
45   @BeforeAll
46   static void sendAndLoadStoredData(
47       @Autowired KafkaTemplate<String, String> messageTemplate,
48       @Autowired ConsumerTaskRunner consumerTaskRunner)
49   {
50     KafkaTestUtils.sendAndLoadStoredData(
51         messageTemplate,
52         INFO_TOPIC,
53         DATA_TOPIC,
54         consumerTaskRunner);
55   }
56
57   @AfterAll
58   static void joinConsumerTasks(
59       @Autowired ConsumerTaskRunner consumerTaskRunner)
60       throws InterruptedException
61   {
62     KafkaTestUtils.joinConsumerTasks(consumerTaskRunner);
63   }
64 }