X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Ftest%2Fjava%2Fde%2Fjuplo%2Fkafka%2Fchat%2Fbackend%2Fpersistence%2Fkafka%2FKafkaServicesConfiguration.java;h=b0e777612fecd6fe59c2134ca0658b853fc3df0e;hb=6f8bd42a20085f186a27d2aeac8aedaf05a7e746;hp=b11babc7ffda058ce7b84c9303ed3837ab532967;hpb=fbbf91af3d2edf8e15b3322d9be300338fc2ed2f;p=demos%2Fkafka%2Fchat diff --git a/src/test/java/de/juplo/kafka/chat/backend/persistence/kafka/KafkaServicesConfiguration.java b/src/test/java/de/juplo/kafka/chat/backend/persistence/kafka/KafkaServicesConfiguration.java index b11babc7..b0e77761 100644 --- a/src/test/java/de/juplo/kafka/chat/backend/persistence/kafka/KafkaServicesConfiguration.java +++ b/src/test/java/de/juplo/kafka/chat/backend/persistence/kafka/KafkaServicesConfiguration.java @@ -4,31 +4,27 @@ import de.juplo.kafka.chat.backend.ChatBackendProperties; import de.juplo.kafka.chat.backend.domain.ChatHome; import de.juplo.kafka.chat.backend.domain.ShardingStrategy; import de.juplo.kafka.chat.backend.persistence.KafkaLikeShardingStrategy; -import jakarta.annotation.PreDestroy; -import lombok.extern.slf4j.Slf4j; import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.common.serialization.IntegerDeserializer; import org.apache.kafka.common.serialization.IntegerSerializer; import org.apache.kafka.common.serialization.StringDeserializer; import org.apache.kafka.common.serialization.StringSerializer; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.ApplicationArguments; -import org.springframework.boot.ApplicationRunner; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; -import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.kafka.support.serializer.JsonDeserializer; import org.springframework.kafka.support.serializer.JsonSerializer; -import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import java.time.Clock; import java.time.ZoneId; +import java.util.HashMap; +import java.util.Map; import java.util.Properties; -import java.util.concurrent.CompletableFuture; @ConditionalOnProperty( @@ -36,61 +32,8 @@ import java.util.concurrent.CompletableFuture; name = "services", havingValue = "kafka") @Configuration -@Slf4j -public class KafkaServicesConfiguration implements ApplicationRunner +public class KafkaServicesConfiguration { - @Autowired - ThreadPoolTaskExecutor taskExecutor; - @Autowired - ConfigurableApplicationContext context; - - @Autowired - ChatMessageChannel chatMessageChannel; - @Autowired - ChatRoomChannel chatRoomChannel; - - CompletableFuture chatRoomChannelConsumerJob; - CompletableFuture chatMessageChannelConsumerJob; - - - @Override - public void run(ApplicationArguments args) throws Exception - { - log.info("Starting the consumer for the ChatRoomChannel"); - chatRoomChannelConsumerJob = taskExecutor - .submitCompletable(chatRoomChannel) - .exceptionally(e -> - { - log.error("The consumer for the ChatRoomChannel exited abnormally!", e); - return null; - }); - log.info("Starting the consumer for the ChatMessageChannel"); - chatMessageChannelConsumerJob = taskExecutor - .submitCompletable(chatMessageChannel) - .exceptionally(e -> - { - log.error("The consumer for the ChatMessageChannel exited abnormally!", e); - return null; - }); - } - - @PreDestroy - public void joinChatRoomChannelConsumerJob() - { - log.info("Waiting for the consumer of the ChatRoomChannel to finish its work"); - chatRoomChannelConsumerJob.join(); - log.info("Joined the consumer of the ChatRoomChannel"); - } - - @PreDestroy - public void joinChatMessageChannelConsumerJob() - { - log.info("Waiting for the consumer of the ChatMessageChannel to finish its work"); - chatMessageChannelConsumerJob.join(); - log.info("Joined the consumer of the ChatMessageChannel"); - } - - @Bean ChatHome kafkaChatHome( ShardingStrategy shardingStrategy, @@ -126,12 +69,18 @@ public class KafkaServicesConfiguration implements ApplicationRunner @Bean Producer chatRoomChannelProducer( - Properties producerProperties, + Properties defaultProducerProperties, + ChatBackendProperties chatBackendProperties, IntegerSerializer integerSerializer, JsonSerializer chatRoomSerializer) { + Map properties = new HashMap<>(); + defaultProducerProperties.forEach((key, value) -> properties.put(key.toString(), value)); + properties.put( + ProducerConfig.CLIENT_ID_CONFIG, + chatBackendProperties.getKafka().getClientIdPrefix() + "_CHATROOM_CHANNEL_PRODUCER"); return new KafkaProducer<>( - producerProperties, + properties, integerSerializer, chatRoomSerializer); } @@ -146,17 +95,29 @@ public class KafkaServicesConfiguration implements ApplicationRunner JsonSerializer chatRoomSerializer() { JsonSerializer serializer = new JsonSerializer<>(); + serializer.configure( + Map.of(JsonSerializer.ADD_TYPE_INFO_HEADERS, false), + false); return serializer; } @Bean Consumer chatRoomChannelConsumer( - Properties producerProperties, + Properties defaultConsumerProperties, + ChatBackendProperties chatBackendProperties, IntegerDeserializer integerDeserializer, JsonDeserializer chatRoomDeserializer) { + Map properties = new HashMap<>(); + defaultConsumerProperties.forEach((key, value) -> properties.put(key.toString(), value)); + properties.put( + ConsumerConfig.CLIENT_ID_CONFIG, + chatBackendProperties.getKafka().getClientIdPrefix() + "_MESSAGE_CHANNEL_CONSUMER"); + properties.put( + ConsumerConfig.GROUP_ID_CONFIG, + "chat_room_channel"); return new KafkaConsumer<>( - producerProperties, + properties, integerDeserializer, chatRoomDeserializer); } @@ -171,6 +132,12 @@ public class KafkaServicesConfiguration implements ApplicationRunner JsonDeserializer chatRoomDeserializer() { JsonDeserializer deserializer = new JsonDeserializer<>(); + deserializer.configure( + Map.of( + JsonDeserializer.USE_TYPE_INFO_HEADERS, false, + JsonDeserializer.VALUE_DEFAULT_TYPE, ChatRoomTo.class, + JsonDeserializer.TRUSTED_PACKAGES, getClass().getPackageName()), + false ); return deserializer; } @@ -197,12 +164,18 @@ public class KafkaServicesConfiguration implements ApplicationRunner @Bean Producer chatMessageChannelProducer( - Properties producerProperties, + Properties defaultProducerProperties, + ChatBackendProperties chatBackendProperties, StringSerializer stringSerializer, JsonSerializer messageSerializer) { + Map properties = new HashMap<>(); + defaultProducerProperties.forEach((key, value) -> properties.put(key.toString(), value)); + properties.put( + ProducerConfig.CLIENT_ID_CONFIG, + chatBackendProperties.getKafka().getClientIdPrefix() + "_MESSAGE_CHANNEL_PRODUCER"); return new KafkaProducer<>( - producerProperties, + properties, stringSerializer, messageSerializer); } @@ -217,17 +190,29 @@ public class KafkaServicesConfiguration implements ApplicationRunner JsonSerializer chatMessageSerializer() { JsonSerializer serializer = new JsonSerializer<>(); + serializer.configure( + Map.of(JsonSerializer.ADD_TYPE_INFO_HEADERS, false), + false); return serializer; } @Bean Consumer chatMessageChannelConsumer( - Properties producerProperties, + Properties defaultConsumerProperties, + ChatBackendProperties chatBackendProperties, StringDeserializer stringDeserializer, JsonDeserializer messageDeserializer) { + Map properties = new HashMap<>(); + defaultConsumerProperties.forEach((key, value) -> properties.put(key.toString(), value)); + properties.put( + ConsumerConfig.CLIENT_ID_CONFIG, + chatBackendProperties.getKafka().getClientIdPrefix() + "_MESSAGE_CHANNEL_CONSUMER"); + properties.put( + ConsumerConfig.GROUP_ID_CONFIG, + "chat_message_channel"); return new KafkaConsumer<>( - producerProperties, + properties, stringDeserializer, messageDeserializer); } @@ -242,20 +227,41 @@ public class KafkaServicesConfiguration implements ApplicationRunner JsonDeserializer chatMessageDeserializer() { JsonDeserializer deserializer = new JsonDeserializer<>(); + deserializer.configure( + Map.of( + JsonDeserializer.USE_TYPE_INFO_HEADERS, false, + JsonDeserializer.VALUE_DEFAULT_TYPE, MessageTo.class, + JsonDeserializer.TRUSTED_PACKAGES, getClass().getPackageName()), + false ); return deserializer; } @Bean - Properties producerProperties(ChatBackendProperties chatBackendProperties) + Properties defaultProducerProperties(ChatBackendProperties chatBackendProperties) { Properties properties = new Properties(); + properties.setProperty( + ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, + chatBackendProperties.getKafka().getBootstrapServers()); return properties; } @Bean - Properties consumerProperties(ChatBackendProperties chatBackendProperties) + Properties defaultConsumerProperties(ChatBackendProperties chatBackendProperties) { Properties properties = new Properties(); + properties.setProperty( + ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, + chatBackendProperties.getKafka().getBootstrapServers()); + properties.setProperty( + ConsumerConfig.CLIENT_ID_CONFIG, + chatBackendProperties.getKafka().getClientIdPrefix()); + properties.setProperty( + ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, + "false"); + properties.setProperty( + ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, + "earliest"); return properties; }