package de.juplo.kafka.chat.backend.implementation.kafka;
+import de.juplo.kafka.chat.backend.ChatBackendProperties;
import de.juplo.kafka.chat.backend.domain.ChatHomeServiceWithShardsTest;
import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.common.TopicPartition;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration;
import org.springframework.boot.autoconfigure.task.TaskExecutionAutoConfiguration;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.context.TestConfiguration;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Import;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.test.context.EmbeddedKafka;
+import org.springframework.test.context.TestPropertySource;
+
+import java.time.Clock;
+import java.util.List;
import static de.juplo.kafka.chat.backend.domain.ChatHomeServiceWithShardsTest.NUM_SHARDS;
import static de.juplo.kafka.chat.backend.implementation.kafka.KafkaChatHomeServiceTest.DATA_TOPIC;
@SpringBootTest(
classes = {
- KafkaTestUtils.KafkaTestConfiguration.class,
+ KafkaChatHomeServiceTest.KafkaChatHomeTestConfiguration.class,
KafkaAutoConfiguration.class,
TaskExecutionAutoConfiguration.class,
},
final static String INFO_TOPIC = "KAFKA_CHAT_HOME_TEST_INFO";
final static String DATA_TOPIC = "KAFKA_CHAT_HOME_TEST_DATA";
+
+ @TestConfiguration
+ @EnableConfigurationProperties(ChatBackendProperties.class)
+ @Import(KafkaServicesConfiguration.class)
+ @TestPropertySource(properties = {
+ "spring.main.allow-bean-definition-overriding=true",
+ "chat.backend.services=kafka",
+ "chat.backend.kafka.client-id-PREFIX=TEST",
+ "chat.backend.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
+ "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
+ })
+ static class KafkaChatHomeTestConfiguration
+ {
+ @Bean
+ WorkAssignor dataChannelWorkAssignor(DataChannel dataChannel)
+ {
+ return consumer ->
+ {
+ List<TopicPartition> assignedPartitions =
+ List.of(new TopicPartition(DATA_TOPIC, 2));
+ consumer.assign(assignedPartitions);
+ dataChannel.onPartitionsAssigned(assignedPartitions);
+ };
+ }
+
+ @Bean
+ Clock clock()
+ {
+ return Clock.systemDefaultZone();
+ }
+ }
+
+
@BeforeAll
static void sendAndLoadStoredData(
@Autowired ConsumerTaskRunner consumerTaskRunner,
@Slf4j
public class KafkaTestUtils
{
- @TestConfiguration
- @EnableConfigurationProperties(ChatBackendProperties.class)
- @Import(KafkaServicesConfiguration.class)
- @TestPropertySource(properties = {
- "spring.main.allow-bean-definition-overriding=true",
- "chat.backend.services=kafka",
- "chat.backend.kafka.client-id-PREFIX=TEST",
- "chat.backend.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
- "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
- })
- public static class KafkaTestConfiguration
- {
- @Bean
- public WorkAssignor dataChannelWorkAssignor(DataChannel dataChannel)
- {
- return consumer ->
- {
- List<TopicPartition> assignedPartitions =
- List.of(new TopicPartition(DATA_TOPIC, 2));
- consumer.assign(assignedPartitions);
- dataChannel.onPartitionsAssigned(assignedPartitions);
- };
- }
-
- @Bean
- public Clock clock()
- {
- return Clock.systemDefaultZone();
- }
- }
-
-
static void sendAndLoadStoredData(
ConsumerTaskRunner consumerTaskRunner,
KafkaTemplate<String, String> messageTemplate)