1 package de.juplo.kafka.chat.backend.implementation.kafka;
3 import de.juplo.kafka.chat.backend.ChatBackendProperties;
4 import de.juplo.kafka.chat.backend.domain.ChatHomeServiceWithShardsTest;
5 import lombok.extern.slf4j.Slf4j;
6 import org.apache.kafka.clients.producer.ProducerRecord;
7 import org.apache.kafka.common.TopicPartition;
8 import org.junit.jupiter.api.AfterAll;
9 import org.junit.jupiter.api.BeforeAll;
10 import org.springframework.beans.factory.annotation.Autowired;
11 import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration;
12 import org.springframework.boot.autoconfigure.task.TaskExecutionAutoConfiguration;
13 import org.springframework.boot.context.properties.EnableConfigurationProperties;
14 import org.springframework.boot.test.context.SpringBootTest;
15 import org.springframework.boot.test.context.TestConfiguration;
16 import org.springframework.context.annotation.Bean;
17 import org.springframework.context.annotation.Import;
18 import org.springframework.kafka.core.KafkaTemplate;
19 import org.springframework.kafka.support.SendResult;
20 import org.springframework.kafka.test.context.EmbeddedKafka;
22 import java.time.Clock;
23 import java.util.List;
25 import static de.juplo.kafka.chat.backend.domain.ChatHomeServiceWithShardsTest.NUM_SHARDS;
26 import static de.juplo.kafka.chat.backend.implementation.kafka.KafkaChatHomeServiceTest.TOPIC;
31 KafkaChatHomeServiceTest.KafkaChatHomeTestConfiguration.class,
32 KafkaAutoConfiguration.class,
33 TaskExecutionAutoConfiguration.class,
36 "spring.main.allow-bean-definition-overriding=true",
37 "chat.backend.services=kafka",
38 "chat.backend.kafka.client-id-PREFIX=TEST",
39 "chat.backend.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
40 "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
41 "chat.backend.kafka.chatroom-channel-topic=" + TOPIC,
42 "chat.backend.kafka.num-partitions=" + NUM_SHARDS,
44 @EmbeddedKafka(topics = { TOPIC }, partitions = 10)
46 public class KafkaChatHomeServiceTest extends ChatHomeServiceWithShardsTest
48 final static String TOPIC = "KAFKA_CHAT_HOME_TEST";
52 @EnableConfigurationProperties(ChatBackendProperties.class)
53 @Import(KafkaServicesConfiguration.class)
54 static class KafkaChatHomeTestConfiguration
57 ConsumerTaskExecutor.WorkAssignor workAssignor(
58 ChatRoomChannel chatRoomChannel)
62 List<TopicPartition> assignedPartitions =
63 List.of(new TopicPartition(TOPIC, 2));
64 consumer.assign(assignedPartitions);
65 chatRoomChannel.onPartitionsAssigned(assignedPartitions);
72 return Clock.systemDefaultZone();
78 public static void sendAndLoadStoredData(
79 @Autowired ConsumerTaskExecutor consumerTaskExecutor,
80 @Autowired KafkaTemplate<String, String> messageTemplate)
82 send(messageTemplate, "5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\": \"5c73531c-6fc4-426c-adcb-afc5c140a0f7\", \"shard\": 2, \"name\": \"FOO\" }", "command_create_chatroom");
83 send(messageTemplate,"5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 1, \"user\" : \"peter\", \"text\" : \"Hallo, ich heiße Peter!\" }", "event_chatmessage_received");
84 send(messageTemplate,"5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 1, \"user\" : \"ute\", \"text\" : \"Ich bin Ute...\" }", "event_chatmessage_received");
85 send(messageTemplate,"5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 2, \"user\" : \"peter\", \"text\" : \"Willst du mit mir gehen?\" }", "event_chatmessage_received");
86 send(messageTemplate,"5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 1, \"user\" : \"klaus\", \"text\" : \"Ja? Nein? Vielleicht??\" }", "event_chatmessage_received");
88 consumerTaskExecutor.executeConsumerTask();
91 static void send(KafkaTemplate<String, String> kafkaTemplate, String key, String value, String typeId)
93 ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, key, value);
94 record.headers().add("__TypeId__", typeId.getBytes());
95 SendResult<String, String> result = kafkaTemplate.send(record).join();
100 new TopicPartition(result.getRecordMetadata().topic(), result.getRecordMetadata().partition()));
104 static void joinConsumerJob(@Autowired ConsumerTaskExecutor consumerTaskExecutor)
106 consumerTaskExecutor.joinConsumerTaskJob();