1 package de.juplo.kafka.chat.backend;
3 import de.juplo.kafka.chat.backend.implementation.kafka.*;
4 import lombok.extern.slf4j.Slf4j;
5 import org.apache.kafka.clients.producer.ProducerRecord;
6 import org.apache.kafka.common.TopicPartition;
7 import org.junit.jupiter.api.AfterAll;
8 import org.junit.jupiter.api.BeforeAll;
9 import org.springframework.beans.factory.annotation.Autowired;
10 import org.springframework.boot.context.properties.EnableConfigurationProperties;
11 import org.springframework.boot.test.context.SpringBootTest;
12 import org.springframework.boot.test.context.TestConfiguration;
13 import org.springframework.boot.test.mock.mockito.MockBean;
14 import org.springframework.context.annotation.Bean;
15 import org.springframework.context.annotation.Import;
16 import org.springframework.kafka.core.KafkaTemplate;
17 import org.springframework.kafka.support.SendResult;
18 import org.springframework.kafka.test.context.EmbeddedKafka;
20 import java.util.List;
22 import static de.juplo.kafka.chat.backend.KafkaConfigurationIT.DATA_TOPIC;
23 import static de.juplo.kafka.chat.backend.KafkaConfigurationIT.INFO_TOPIC;
27 webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
29 "spring.main.allow-bean-definition-overriding=true",
30 "chat.backend.services=kafka",
31 "chat.backend.kafka.client-id-PREFIX=TEST",
32 "chat.backend.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
33 "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
34 "chat.backend.kafka.info-channel-topic=" + INFO_TOPIC,
35 "chat.backend.kafka.data-channel-topic=" + DATA_TOPIC,
36 "chat.backend.kafka.num-partitions=10",
39 topics = { INFO_TOPIC, DATA_TOPIC },
42 class KafkaConfigurationIT extends AbstractConfigurationWithShardingIT
44 final static String INFO_TOPIC = "KAFKA_CONFIGURATION_IT_INFO_CHANNEL";
45 final static String DATA_TOPIC = "KAFKA_CONFIGURATION_IT_DATA_CHANNEL";
48 KafkaServicesApplicationRunner kafkaServicesApplicationRunner;
51 public static void sendAndLoadStoredData(
52 @Autowired KafkaTemplate<String, String> messageTemplate,
53 @Autowired ConsumerTaskRunner consumerTaskRunner)
55 send(messageTemplate, INFO_TOPIC, "5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\": \"5c73531c-6fc4-426c-adcb-afc5c140a0f7\", \"shard\": 2, \"name\": \"FOO\" }", "event_chatroom_created");
56 send(messageTemplate, DATA_TOPIC, "5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 1, \"user\" : \"peter\", \"text\" : \"Hallo, ich heiße Peter!\" }", "event_chatmessage_received");
57 send(messageTemplate, DATA_TOPIC, "5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 1, \"user\" : \"ute\", \"text\" : \"Ich bin Ute...\" }", "event_chatmessage_received");
58 send(messageTemplate, DATA_TOPIC, "5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 2, \"user\" : \"peter\", \"text\" : \"Willst du mit mir gehen?\" }", "event_chatmessage_received");
59 send(messageTemplate, DATA_TOPIC, "5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 1, \"user\" : \"klaus\", \"text\" : \"Ja? Nein? Vielleicht??\" }", "event_chatmessage_received");
61 consumerTaskRunner.executeConsumerTasks();
65 KafkaTemplate<String, String> kafkaTemplate,
71 ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);
72 record.headers().add("__TypeId__", typeId.getBytes());
73 SendResult<String, String> result = kafkaTemplate.send(record).join();
78 new TopicPartition(result.getRecordMetadata().topic(), result.getRecordMetadata().partition()));
81 static void joinConsumerTasks(@Autowired ConsumerTaskRunner consumerTaskRunner)
83 consumerTaskRunner.joinConsumerTasks();
88 @EnableConfigurationProperties(ChatBackendProperties.class)
89 @Import(KafkaServicesConfiguration.class)
90 static class KafkaConfigurationITConfiguration
93 ConsumerTaskExecutor.WorkAssignor dataChannelWorkAssignor(
94 DataChannel dataChannel)
98 List<TopicPartition> assignedPartitions =
99 List.of(new TopicPartition(DATA_TOPIC, 2));
100 consumer.assign(assignedPartitions);
101 dataChannel.onPartitionsAssigned(assignedPartitions);