1 package de.juplo.kafka.chat.backend;
3 import de.juplo.kafka.chat.backend.implementation.kafka.ChatRoomChannel;
4 import de.juplo.kafka.chat.backend.implementation.kafka.KafkaServicesApplicationRunner;
5 import lombok.extern.slf4j.Slf4j;
6 import org.apache.kafka.clients.consumer.Consumer;
7 import org.apache.kafka.clients.producer.ProducerRecord;
8 import org.apache.kafka.common.TopicPartition;
9 import org.junit.jupiter.api.AfterAll;
10 import org.junit.jupiter.api.BeforeAll;
11 import org.springframework.beans.factory.annotation.Autowired;
12 import org.springframework.boot.test.context.SpringBootTest;
13 import org.springframework.boot.test.mock.mockito.MockBean;
14 import org.springframework.kafka.core.KafkaTemplate;
15 import org.springframework.kafka.support.SendResult;
16 import org.springframework.kafka.test.context.EmbeddedKafka;
17 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
19 import java.util.List;
20 import java.util.concurrent.CompletableFuture;
22 import static de.juplo.kafka.chat.backend.KafkaConfigurationIT.TOPIC;
26 webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
28 "chat.backend.services=kafka",
29 "chat.backend.kafka.client-id-PREFIX=TEST",
30 "chat.backend.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
31 "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
32 "chat.backend.kafka.chatroom-channel-topic=" + TOPIC,
33 "chat.backend.kafka.num-partitions=10",
35 @EmbeddedKafka(topics = { TOPIC }, partitions = 10)
37 class KafkaConfigurationIT extends AbstractConfigurationWithShardingIT
39 final static String TOPIC = "KAFKA_CONFIGURATION_IT";
41 static CompletableFuture<Void> CONSUMER_JOB;
44 KafkaServicesApplicationRunner kafkaServicesApplicationRunner;
47 public static void sendAndLoadStoredData(
48 @Autowired KafkaTemplate<String, String> messageTemplate,
49 @Autowired Consumer chatRoomChannelConsumer,
50 @Autowired ThreadPoolTaskExecutor taskExecutor,
51 @Autowired ChatRoomChannel chatRoomChannel)
53 send(messageTemplate, "5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\": \"5c73531c-6fc4-426c-adcb-afc5c140a0f7\", \"shard\": 2, \"name\": \"FOO\" }", "command_create_chatroom");
54 send(messageTemplate,"5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 1, \"user\" : \"peter\", \"text\" : \"Hallo, ich heiße Peter!\" }", "event_chatmessage_received");
55 send(messageTemplate,"5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 1, \"user\" : \"ute\", \"text\" : \"Ich bin Ute...\" }", "event_chatmessage_received");
56 send(messageTemplate,"5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 2, \"user\" : \"peter\", \"text\" : \"Willst du mit mir gehen?\" }", "event_chatmessage_received");
57 send(messageTemplate,"5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 1, \"user\" : \"klaus\", \"text\" : \"Ja? Nein? Vielleicht??\" }", "event_chatmessage_received");
59 List<TopicPartition> assignedPartitions = List.of(new TopicPartition(TOPIC, 2));
60 chatRoomChannelConsumer.assign(assignedPartitions);
61 chatRoomChannel.onPartitionsAssigned(assignedPartitions);
62 CONSUMER_JOB = taskExecutor
63 .submitCompletable(chatRoomChannel)
66 log.error("The consumer for the ChatRoomChannel exited abnormally!", e);
71 static void send(KafkaTemplate<String, String> kafkaTemplate, String key, String value, String typeId)
73 ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, key, value);
74 record.headers().add("__TypeId__", typeId.getBytes());
75 SendResult<String, String> result = kafkaTemplate.send(record).join();
80 new TopicPartition(result.getRecordMetadata().topic(), result.getRecordMetadata().partition()));
84 static void joinConsumerJob(@Autowired Consumer chatRoomChannelConsumer)
86 log.info("Signaling the consumer of the CahtRoomChannel to quit its work");
87 chatRoomChannelConsumer.wakeup();
88 log.info("Waiting for the consumer of the ChatRoomChannel to finish its work");
90 log.info("Joined the consumer of the ChatRoomChannel");