1 package de.juplo.kafka.chat.backend.implementation.kafka;
3 import de.juplo.kafka.chat.backend.domain.ChatHomeServiceWithShardsTest;
4 import lombok.extern.slf4j.Slf4j;
5 import org.apache.kafka.clients.consumer.Consumer;
6 import org.junit.jupiter.api.AfterAll;
7 import org.junit.jupiter.api.BeforeAll;
8 import org.springframework.beans.factory.annotation.Autowired;
9 import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration;
10 import org.springframework.boot.autoconfigure.task.TaskExecutionAutoConfiguration;
11 import org.springframework.kafka.core.KafkaTemplate;
12 import org.springframework.kafka.test.context.EmbeddedKafka;
13 import org.springframework.test.context.ContextConfiguration;
14 import org.springframework.test.context.TestPropertySource;
16 import static de.juplo.kafka.chat.backend.domain.ChatHomeServiceWithShardsTest.NUM_SHARDS;
17 import static de.juplo.kafka.chat.backend.implementation.kafka.KafkaChatHomeServiceTest.DATA_TOPIC;
18 import static de.juplo.kafka.chat.backend.implementation.kafka.KafkaChatHomeServiceTest.INFO_TOPIC;
21 @ContextConfiguration(classes = {
22 KafkaTestUtils.KafkaTestConfiguration.class,
23 KafkaAutoConfiguration.class,
24 TaskExecutionAutoConfiguration.class,
26 @TestPropertySource(properties = {
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,
36 topics = { INFO_TOPIC, DATA_TOPIC },
37 partitions = NUM_SHARDS)
39 public class KafkaChatHomeServiceTest extends ChatHomeServiceWithShardsTest
41 final static String INFO_TOPIC = "KAFKA_CHAT_HOME_TEST_INFO";
42 final static String DATA_TOPIC = "KAFKA_CHAT_HOME_TEST_DATA";
46 static void sendAndLoadStoredData(
47 @Autowired KafkaTemplate<String, String> messageTemplate,
48 @Autowired ChannelTaskRunner channelTaskRunner)
50 KafkaTestUtils.sendAndLoadStoredData(
58 static void joinChannels(
59 @Autowired Consumer dataChannelConsumer,
60 @Autowired Consumer infoChannelConsumer,
61 @Autowired ChannelTaskRunner channelTaskRunner)
62 throws InterruptedException
64 dataChannelConsumer.wakeup();
65 infoChannelConsumer.wakeup();
66 channelTaskRunner.joinChannels();