diff --git a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaAnnotationDrivenConfiguration.java b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaAnnotationDrivenConfiguration.java index abbc834f46..94a746901d 100644 --- a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaAnnotationDrivenConfiguration.java +++ b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaAnnotationDrivenConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2012-2023 the original author or authors. + * Copyright 2012-2025 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -23,7 +23,6 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnThreading; import org.springframework.boot.autoconfigure.thread.Threading; -import org.springframework.boot.ssl.SslBundles; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.core.task.SimpleAsyncTaskExecutor; @@ -152,11 +151,10 @@ class KafkaAnnotationDrivenConfiguration { ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory( ConcurrentKafkaListenerContainerFactoryConfigurer configurer, ObjectProvider> kafkaConsumerFactory, - ObjectProvider>> kafkaContainerCustomizer, - ObjectProvider sslBundles) { + ObjectProvider>> kafkaContainerCustomizer) { ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory<>(); - configurer.configure(factory, kafkaConsumerFactory.getIfAvailable(() -> new DefaultKafkaConsumerFactory<>( - this.properties.buildConsumerProperties(sslBundles.getIfAvailable())))); + configurer.configure(factory, kafkaConsumerFactory + .getIfAvailable(() -> new DefaultKafkaConsumerFactory<>(this.properties.buildConsumerProperties()))); kafkaContainerCustomizer.ifAvailable(factory::setContainerCustomizer); return factory; } diff --git a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfiguration.java b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfiguration.java index 3c949d9dbc..ba0968b9f9 100644 --- a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfiguration.java +++ b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfiguration.java @@ -23,6 +23,7 @@ import java.util.Map; import org.apache.kafka.clients.CommonClientConfigs; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.common.config.SslConfigs; import org.springframework.beans.factory.ObjectProvider; import org.springframework.boot.autoconfigure.AutoConfiguration; @@ -32,10 +33,12 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.autoconfigure.condition.ConditionalOnSingleCandidate; +import org.springframework.boot.autoconfigure.kafka.KafkaConnectionDetails.Configuration; import org.springframework.boot.autoconfigure.kafka.KafkaProperties.Jaas; import org.springframework.boot.autoconfigure.kafka.KafkaProperties.Retry.Topic.Backoff; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.boot.context.properties.PropertyMapper; +import org.springframework.boot.ssl.SslBundle; import org.springframework.boot.ssl.SslBundles; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Import; @@ -54,6 +57,7 @@ import org.springframework.kafka.support.converter.RecordMessageConverter; import org.springframework.kafka.transaction.KafkaTransactionManager; import org.springframework.retry.backoff.BackOffPolicyBuilder; import org.springframework.retry.backoff.SleepingBackOffPolicy; +import org.springframework.util.StringUtils; /** * {@link EnableAutoConfiguration Auto-configuration} for Apache Kafka. @@ -84,8 +88,9 @@ public class KafkaAutoConfiguration { @Bean @ConditionalOnMissingBean(KafkaConnectionDetails.class) - PropertiesKafkaConnectionDetails kafkaConnectionDetails(KafkaProperties properties) { - return new PropertiesKafkaConnectionDetails(properties); + PropertiesKafkaConnectionDetails kafkaConnectionDetails(KafkaProperties properties, + ObjectProvider sslBundles) { + return new PropertiesKafkaConnectionDetails(properties, sslBundles.getIfAvailable()); } @Bean @@ -111,9 +116,9 @@ public class KafkaAutoConfiguration { @Bean @ConditionalOnMissingBean(ConsumerFactory.class) - public DefaultKafkaConsumerFactory kafkaConsumerFactory(KafkaConnectionDetails connectionDetails, - ObjectProvider customizers, ObjectProvider sslBundles) { - Map properties = this.properties.buildConsumerProperties(sslBundles.getIfAvailable()); + DefaultKafkaConsumerFactory kafkaConsumerFactory(KafkaConnectionDetails connectionDetails, + ObjectProvider customizers) { + Map properties = this.properties.buildConsumerProperties(); applyKafkaConnectionDetailsForConsumer(properties, connectionDetails); DefaultKafkaConsumerFactory factory = new DefaultKafkaConsumerFactory<>(properties); customizers.orderedStream().forEach((customizer) -> customizer.customize(factory)); @@ -122,9 +127,9 @@ public class KafkaAutoConfiguration { @Bean @ConditionalOnMissingBean(ProducerFactory.class) - public DefaultKafkaProducerFactory kafkaProducerFactory(KafkaConnectionDetails connectionDetails, - ObjectProvider customizers, ObjectProvider sslBundles) { - Map properties = this.properties.buildProducerProperties(sslBundles.getIfAvailable()); + DefaultKafkaProducerFactory kafkaProducerFactory(KafkaConnectionDetails connectionDetails, + ObjectProvider customizers) { + Map properties = this.properties.buildProducerProperties(); applyKafkaConnectionDetailsForProducer(properties, connectionDetails); DefaultKafkaProducerFactory factory = new DefaultKafkaProducerFactory<>(properties); String transactionIdPrefix = this.properties.getProducer().getTransactionIdPrefix(); @@ -160,8 +165,8 @@ public class KafkaAutoConfiguration { @Bean @ConditionalOnMissingBean - public KafkaAdmin kafkaAdmin(KafkaConnectionDetails connectionDetails, ObjectProvider sslBundles) { - Map properties = this.properties.buildAdminProperties(sslBundles.getIfAvailable()); + KafkaAdmin kafkaAdmin(KafkaConnectionDetails connectionDetails) { + Map properties = this.properties.buildAdminProperties(null); applyKafkaConnectionDetailsForAdmin(properties, connectionDetails); KafkaAdmin kafkaAdmin = new KafkaAdmin(properties); KafkaProperties.Admin admin = this.properties.getAdmin(); @@ -193,26 +198,26 @@ public class KafkaAutoConfiguration { private void applyKafkaConnectionDetailsForConsumer(Map properties, KafkaConnectionDetails connectionDetails) { - properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, connectionDetails.getConsumerBootstrapServers()); - if (!(connectionDetails instanceof PropertiesKafkaConnectionDetails)) { - properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "PLAINTEXT"); - } + Configuration consumer = connectionDetails.getConsumer(); + properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, consumer.getBootstrapServers()); + applySecurityProtocol(properties, connectionDetails.getSecurityProtocol()); + applySslBundle(properties, consumer.getSslBundle()); } private void applyKafkaConnectionDetailsForProducer(Map properties, KafkaConnectionDetails connectionDetails) { - properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, connectionDetails.getProducerBootstrapServers()); - if (!(connectionDetails instanceof PropertiesKafkaConnectionDetails)) { - properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "PLAINTEXT"); - } + Configuration producer = connectionDetails.getProducer(); + properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, producer.getBootstrapServers()); + applySecurityProtocol(properties, producer.getSecurityProtocol()); + applySslBundle(properties, producer.getSslBundle()); } private void applyKafkaConnectionDetailsForAdmin(Map properties, KafkaConnectionDetails connectionDetails) { - properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, connectionDetails.getAdminBootstrapServers()); - if (!(connectionDetails instanceof PropertiesKafkaConnectionDetails)) { - properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "PLAINTEXT"); - } + Configuration admin = connectionDetails.getAdmin(); + properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, admin.getBootstrapServers()); + applySecurityProtocol(properties, admin.getSecurityProtocol()); + applySslBundle(properties, admin.getSslBundle()); } private static void setBackOffPolicy(RetryTopicConfigurationBuilder builder, Backoff retryTopicBackoff) { @@ -231,4 +236,17 @@ public class KafkaAutoConfiguration { } } + static void applySslBundle(Map properties, SslBundle sslBundle) { + if (sslBundle != null) { + properties.put(SslConfigs.SSL_ENGINE_FACTORY_CLASS_CONFIG, SslBundleSslEngineFactory.class.getName()); + properties.put(SslBundle.class.getName(), sslBundle); + } + } + + static void applySecurityProtocol(Map properties, String securityProtocol) { + if (StringUtils.hasLength(securityProtocol)) { + properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, securityProtocol); + } + } + } diff --git a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaConnectionDetails.java b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaConnectionDetails.java index 5490e9852e..19214b4c5a 100644 --- a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaConnectionDetails.java +++ b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaConnectionDetails.java @@ -1,5 +1,5 @@ /* - * Copyright 2012-2023 the original author or authors. + * Copyright 2012-2025 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -19,6 +19,7 @@ package org.springframework.boot.autoconfigure.kafka; import java.util.List; import org.springframework.boot.autoconfigure.service.connection.ConnectionDetails; +import org.springframework.boot.ssl.SslBundle; /** * Details required to establish a connection to a Kafka service. @@ -36,36 +37,173 @@ public interface KafkaConnectionDetails extends ConnectionDetails { */ List getBootstrapServers(); + /** + * Returns the SSL bundle. + * @return the SSL bundle + * @since 3.5.0 + */ + default SslBundle getSslBundle() { + return null; + } + + /** + * Returns the security protocol. + * @return the security protocol + * @since 3.5.0 + */ + default String getSecurityProtocol() { + return null; + } + + /** + * Returns the consumer configuration. + * @return the consumer configuration + * @since 3.5.0 + */ + default Configuration getConsumer() { + return Configuration.of(getBootstrapServers(), getSslBundle(), getSecurityProtocol()); + } + + /** + * Returns the producer configuration. + * @return the producer configuration + * @since 3.5.0 + */ + default Configuration getProducer() { + return Configuration.of(getBootstrapServers(), getSslBundle(), getSecurityProtocol()); + } + + /** + * Returns the admin configuration. + * @return the admin configuration + * @since 3.5.0 + */ + default Configuration getAdmin() { + return Configuration.of(getBootstrapServers(), getSslBundle(), getSecurityProtocol()); + } + + /** + * Returns the Kafka Streams configuration. + * @return the Kafka Streams configuration + * @since 3.5.0 + */ + default Configuration getStreams() { + return Configuration.of(getBootstrapServers(), getSslBundle(), getSecurityProtocol()); + } + /** * Returns the list of bootstrap servers used for consumers. * @return the list of bootstrap servers used for consumers + * @deprecated since 3.5.0 for removal in 3.7.0 in favor of {@link #getConsumer()} */ + @Deprecated(since = "3.5.0", forRemoval = true) default List getConsumerBootstrapServers() { - return getBootstrapServers(); + return getConsumer().getBootstrapServers(); } /** * Returns the list of bootstrap servers used for producers. * @return the list of bootstrap servers used for producers + * @deprecated since 3.5.0 for removal in 3.7.0 in favor of {@link #getProducer()} */ + @Deprecated(since = "3.5.0", forRemoval = true) default List getProducerBootstrapServers() { - return getBootstrapServers(); + return getProducer().getBootstrapServers(); } /** * Returns the list of bootstrap servers used for the admin. * @return the list of bootstrap servers used for the admin + * @deprecated since 3.5.0 for removal in 3.7.0 in favor of {@link #getAdmin()} */ + @Deprecated(since = "3.5.0", forRemoval = true) default List getAdminBootstrapServers() { - return getBootstrapServers(); + return getAdmin().getBootstrapServers(); } /** * Returns the list of bootstrap servers used for Kafka Streams. * @return the list of bootstrap servers used for Kafka Streams + * @deprecated since 3.5.0 for removal in 3.7.0 in favor of {@link #getStreams()} */ + @Deprecated(since = "3.5.0", forRemoval = true) default List getStreamsBootstrapServers() { - return getBootstrapServers(); + return getStreams().getBootstrapServers(); + } + + /** + * Kafka connection details configuration. + */ + interface Configuration { + + /** + * Creates a new configuration with the given bootstrap servers. + * @param bootstrapServers the bootstrap servers + * @return the configuration + */ + static Configuration of(List bootstrapServers) { + return Configuration.of(bootstrapServers, null, null); + } + + /** + * Creates a new configuration with the given bootstrap servers and SSL bundle. + * @param bootstrapServers the bootstrap servers + * @param sslBundle the SSL bundle + * @return the configuration + */ + static Configuration of(List bootstrapServers, SslBundle sslBundle) { + return Configuration.of(bootstrapServers, sslBundle, null); + } + + /** + * Creates a new configuration with the given bootstrap servers, SSL bundle and + * security protocol. + * @param bootstrapServers the bootstrap servers + * @param sslBundle the SSL bundle + * @param securityProtocol the security protocol + * @return the configuration + */ + static Configuration of(List bootstrapServers, SslBundle sslBundle, String securityProtocol) { + return new Configuration() { + @Override + public List getBootstrapServers() { + return bootstrapServers; + } + + @Override + public SslBundle getSslBundle() { + return sslBundle; + } + + @Override + public String getSecurityProtocol() { + return securityProtocol; + } + }; + } + + /** + * Returns the list of bootstrap servers. + * @return the list of bootstrap servers + */ + List getBootstrapServers(); + + /** + * Returns the SSL bundle. + * @return the SSL bundle + */ + default SslBundle getSslBundle() { + return null; + } + + /** + * Returns the security protocol. + * @return the security protocol + */ + default String getSecurityProtocol() { + return null; + } + } } diff --git a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaProperties.java b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaProperties.java index b70813bb5e..ae6cc4393e 100644 --- a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaProperties.java +++ b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaProperties.java @@ -38,7 +38,6 @@ import org.springframework.boot.context.properties.DeprecatedConfigurationProper import org.springframework.boot.context.properties.PropertyMapper; import org.springframework.boot.context.properties.source.MutuallyExclusiveConfigurationPropertiesException; import org.springframework.boot.convert.DurationUnit; -import org.springframework.boot.ssl.SslBundle; import org.springframework.boot.ssl.SslBundles; import org.springframework.core.io.Resource; import org.springframework.kafka.listener.ContainerProperties.AckMode; @@ -1401,10 +1400,10 @@ public class KafkaProperties { public Map buildProperties(SslBundles sslBundles) { validate(); String bundleName = getBundle(); - if (StringUtils.hasText(bundleName)) { - return buildPropertiesForSslBundle(sslBundles, bundleName); - } Properties properties = new Properties(); + if (StringUtils.hasText(bundleName)) { + return properties; + } PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull(); map.from(this::getKeyPassword).to(properties.in(SslConfigs.SSL_KEY_PASSWORD_CONFIG)); map.from(this::getKeyStoreCertificateChain) @@ -1425,13 +1424,6 @@ public class KafkaProperties { return properties; } - private Map buildPropertiesForSslBundle(SslBundles sslBundles, String name) { - Properties properties = new Properties(); - properties.in(SslConfigs.SSL_ENGINE_FACTORY_CLASS_CONFIG).accept(SslBundleSslEngineFactory.class.getName()); - properties.in(SslBundle.class.getName()).accept(sslBundles.getBundle(name)); - return properties; - } - private void validate() { MutuallyExclusiveConfigurationPropertiesException.throwIfMultipleMatchingValuesIn((entries) -> { entries.put("spring.kafka.ssl.key-store-key", getKeyStoreKey()); diff --git a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaStreamsAnnotationDrivenConfiguration.java b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaStreamsAnnotationDrivenConfiguration.java index 7013843263..a3078a0dad 100644 --- a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaStreamsAnnotationDrivenConfiguration.java +++ b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaStreamsAnnotationDrivenConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2012-2023 the original author or authors. + * Copyright 2012-2025 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -18,7 +18,6 @@ package org.springframework.boot.autoconfigure.kafka; import java.util.Map; -import org.apache.kafka.clients.CommonClientConfigs; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.streams.StreamsBuilder; import org.apache.kafka.streams.StreamsConfig; @@ -30,7 +29,6 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.context.properties.source.InvalidConfigurationPropertyValueException; -import org.springframework.boot.ssl.SslBundles; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.core.env.Environment; @@ -63,8 +61,8 @@ class KafkaStreamsAnnotationDrivenConfiguration { @ConditionalOnMissingBean @Bean(KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME) KafkaStreamsConfiguration defaultKafkaStreamsConfig(Environment environment, - KafkaConnectionDetails connectionDetails, ObjectProvider sslBundles) { - Map properties = this.properties.buildStreamsProperties(sslBundles.getIfAvailable()); + KafkaConnectionDetails connectionDetails) { + Map properties = this.properties.buildStreamsProperties(null); applyKafkaConnectionDetailsForStreams(properties, connectionDetails); if (this.properties.getStreams().getApplicationId() == null) { String applicationName = environment.getProperty("spring.application.name"); @@ -87,10 +85,10 @@ class KafkaStreamsAnnotationDrivenConfiguration { private void applyKafkaConnectionDetailsForStreams(Map properties, KafkaConnectionDetails connectionDetails) { - properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, connectionDetails.getStreamsBootstrapServers()); - if (!(connectionDetails instanceof PropertiesKafkaConnectionDetails)) { - properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "PLAINTEXT"); - } + KafkaConnectionDetails.Configuration streams = connectionDetails.getStreams(); + properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, streams.getBootstrapServers()); + KafkaAutoConfiguration.applySecurityProtocol(properties, streams.getSecurityProtocol()); + KafkaAutoConfiguration.applySslBundle(properties, streams.getSslBundle()); } // Separate class required to avoid BeanCurrentlyInCreationException diff --git a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/PropertiesKafkaConnectionDetails.java b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/PropertiesKafkaConnectionDetails.java index 26ea6ad67f..310e107aaf 100644 --- a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/PropertiesKafkaConnectionDetails.java +++ b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/PropertiesKafkaConnectionDetails.java @@ -1,5 +1,5 @@ /* - * Copyright 2012-2023 the original author or authors. + * Copyright 2012-2025 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -18,6 +18,12 @@ package org.springframework.boot.autoconfigure.kafka; import java.util.List; +import org.springframework.boot.autoconfigure.kafka.KafkaProperties.Ssl; +import org.springframework.boot.ssl.SslBundle; +import org.springframework.boot.ssl.SslBundles; +import org.springframework.util.Assert; +import org.springframework.util.StringUtils; + /** * Adapts {@link KafkaProperties} to {@link KafkaConnectionDetails}. * @@ -29,8 +35,11 @@ class PropertiesKafkaConnectionDetails implements KafkaConnectionDetails { private final KafkaProperties properties; - PropertiesKafkaConnectionDetails(KafkaProperties properties) { + private final SslBundles sslBundles; + + PropertiesKafkaConnectionDetails(KafkaProperties properties, SslBundles sslBundles) { this.properties = properties; + this.sslBundles = sslBundles; } @Override @@ -39,22 +48,59 @@ class PropertiesKafkaConnectionDetails implements KafkaConnectionDetails { } @Override - public List getConsumerBootstrapServers() { - return getServers(this.properties.getConsumer().getBootstrapServers()); + public Configuration getConsumer() { + List servers = this.properties.getConsumer().getBootstrapServers(); + SslBundle sslBundle = getBundle(this.properties.getConsumer().getSsl()); + String protocol = this.properties.getConsumer().getSecurity().getProtocol(); + return Configuration.of((servers != null) ? servers : getBootstrapServers(), + (sslBundle != null) ? sslBundle : getSslBundle(), + (StringUtils.hasLength(protocol)) ? protocol : getSecurityProtocol()); } @Override - public List getProducerBootstrapServers() { - return getServers(this.properties.getProducer().getBootstrapServers()); + public Configuration getProducer() { + List servers = this.properties.getProducer().getBootstrapServers(); + SslBundle sslBundle = getBundle(this.properties.getProducer().getSsl()); + String protocol = this.properties.getProducer().getSecurity().getProtocol(); + return Configuration.of((servers != null) ? servers : getBootstrapServers(), + (sslBundle != null) ? sslBundle : getSslBundle(), + (StringUtils.hasLength(protocol)) ? protocol : getSecurityProtocol()); } @Override - public List getStreamsBootstrapServers() { - return getServers(this.properties.getStreams().getBootstrapServers()); + public Configuration getStreams() { + List servers = this.properties.getStreams().getBootstrapServers(); + SslBundle sslBundle = getBundle(this.properties.getStreams().getSsl()); + String protocol = this.properties.getStreams().getSecurity().getProtocol(); + return Configuration.of((servers != null) ? servers : getBootstrapServers(), + (sslBundle != null) ? sslBundle : getSslBundle(), + (StringUtils.hasLength(protocol)) ? protocol : getSecurityProtocol()); } - private List getServers(List servers) { - return (servers != null) ? servers : getBootstrapServers(); + @Override + public Configuration getAdmin() { + SslBundle sslBundle = getBundle(this.properties.getAdmin().getSsl()); + String protocol = this.properties.getAdmin().getSecurity().getProtocol(); + return Configuration.of(getBootstrapServers(), (sslBundle != null) ? sslBundle : getSslBundle(), + (StringUtils.hasLength(protocol)) ? protocol : getSecurityProtocol()); + } + + @Override + public SslBundle getSslBundle() { + return getBundle(this.properties.getSsl()); + } + + @Override + public String getSecurityProtocol() { + return this.properties.getSecurity().getProtocol(); + } + + private SslBundle getBundle(Ssl ssl) { + if (StringUtils.hasLength(ssl.getBundle())) { + Assert.notNull(this.sslBundles, "SSL bundle name has been set but no SSL bundles found in context"); + return this.sslBundles.getBundle(ssl.getBundle()); + } + return null; } } diff --git a/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfigurationTests.java b/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfigurationTests.java index 8e34ee8da5..add0f3f073 100644 --- a/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfigurationTests.java +++ b/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfigurationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2012-2024 the original author or authors. + * Copyright 2012-2025 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -47,6 +47,8 @@ import org.junit.jupiter.params.provider.ValueSource; import org.springframework.boot.autoconfigure.AutoConfigurations; import org.springframework.boot.autoconfigure.ssl.SslAutoConfiguration; +import org.springframework.boot.ssl.SslBundle; +import org.springframework.boot.ssl.SslStoreBundle; import org.springframework.boot.test.context.assertj.AssertableApplicationContext; import org.springframework.boot.test.context.runner.ApplicationContextRunner; import org.springframework.boot.test.context.runner.ContextConsumer; @@ -192,10 +194,34 @@ class KafkaAutoConfigurationTests { Collections.singletonList("kafka.example.com:12345")); assertThat(configs).containsEntry(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, Collections.singletonList("kafka.example.com:12345")); - assertThat(configs).containsEntry(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "PLAINTEXT"); }); } + @Test + void connectionDetailsWithSslBundleAreAppliedToConsumer() { + SslBundle sslBundle = SslBundle.of(SslStoreBundle.NONE); + KafkaConnectionDetails connectionDetails = new KafkaConnectionDetails() { + @Override + public List getBootstrapServers() { + return List.of("kafka.example.com:12345"); + } + + @Override + public Configuration getConsumer() { + return Configuration.of(getBootstrapServers(), sslBundle); + } + + }; + this.contextRunner.withBean(KafkaConnectionDetails.class, () -> connectionDetails).run((context) -> { + assertThat(context).hasSingleBean(KafkaConnectionDetails.class); + DefaultKafkaConsumerFactory consumerFactory = context.getBean(DefaultKafkaConsumerFactory.class); + Map configs = consumerFactory.getConfigurationProperties(); + assertThat(configs).containsEntry("ssl.engine.factory.class", + "org.springframework.boot.autoconfigure.kafka.SslBundleSslEngineFactory"); + assertThat(configs).containsEntry("org.springframework.boot.ssl.SslBundle", sslBundle); + }); + } + @Test void producerProperties() { this.contextRunner.withPropertyValues("spring.kafka.clientId=cid", @@ -262,10 +288,34 @@ class KafkaAutoConfigurationTests { Collections.singletonList("kafka.example.com:12345")); assertThat(configs).containsEntry(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, Collections.singletonList("kafka.example.com:12345")); - assertThat(configs).containsEntry(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "PLAINTEXT"); }); } + @Test + void connectionDetailsWithSslBundleAreAppliedToProducer() { + SslBundle sslBundle = SslBundle.of(SslStoreBundle.NONE); + KafkaConnectionDetails connectionDetails = new KafkaConnectionDetails() { + @Override + public List getBootstrapServers() { + return List.of("kafka.example.com:12345"); + } + + @Override + public Configuration getProducer() { + return Configuration.of(getBootstrapServers(), sslBundle); + } + + }; + this.contextRunner.withBean(KafkaConnectionDetails.class, () -> connectionDetails).run((context) -> { + assertThat(context).hasSingleBean(KafkaConnectionDetails.class); + DefaultKafkaProducerFactory producerFactory = context.getBean(DefaultKafkaProducerFactory.class); + Map configs = producerFactory.getConfigurationProperties(); + assertThat(configs).containsEntry("ssl.engine.factory.class", + "org.springframework.boot.autoconfigure.kafka.SslBundleSslEngineFactory"); + assertThat(configs).containsEntry("org.springframework.boot.ssl.SslBundle", sslBundle); + }); + } + @Test void adminProperties() { this.contextRunner @@ -322,11 +372,34 @@ class KafkaAutoConfigurationTests { Collections.singletonList("kafka.example.com:12345")); assertThat(configs).containsEntry(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, Collections.singletonList("kafka.example.com:12345")); - assertThat(configs).containsEntry(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "PLAINTEXT"); - assertThat(configs).containsEntry(AdminClientConfig.SECURITY_PROTOCOL_CONFIG, "PLAINTEXT"); }); } + @Test + void connectionDetailsWithSslBundleAreAppliedToAdmin() { + SslBundle sslBundle = SslBundle.of(SslStoreBundle.NONE); + KafkaConnectionDetails connectionDetails = new KafkaConnectionDetails() { + @Override + public List getBootstrapServers() { + return List.of("kafka.example.com:12345"); + } + + @Override + public Configuration getAdmin() { + return Configuration.of(getBootstrapServers(), sslBundle); + } + + }; + this.contextRunner.withBean(KafkaConnectionDetails.class, () -> connectionDetails).run((context) -> { + assertThat(context).hasSingleBean(KafkaConnectionDetails.class); + KafkaAdmin admin = context.getBean(KafkaAdmin.class); + Map configs = admin.getConfigurationProperties(); + assertThat(configs).containsEntry("ssl.engine.factory.class", + "org.springframework.boot.autoconfigure.kafka.SslBundleSslEngineFactory"); + assertThat(configs).containsEntry("org.springframework.boot.ssl.SslBundle", sslBundle); + }); + } + @SuppressWarnings("unchecked") @Test void streamsProperties() { @@ -391,8 +464,35 @@ class KafkaAutoConfigurationTests { Collections.singletonList("kafka.example.com:12345")); assertThat(configs).containsEntry(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, Collections.singletonList("kafka.example.com:12345")); - assertThat(configs).containsEntry(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "PLAINTEXT"); - assertThat(configs).containsEntry(StreamsConfig.SECURITY_PROTOCOL_CONFIG, "PLAINTEXT"); + }); + } + + @Test + void connectionDetailsWithSslBundleAreAppliedToStreams() { + SslBundle sslBundle = SslBundle.of(SslStoreBundle.NONE); + KafkaConnectionDetails connectionDetails = new KafkaConnectionDetails() { + @Override + public List getBootstrapServers() { + return List.of("kafka.example.com:12345"); + } + + @Override + public Configuration getStreams() { + return Configuration.of(getBootstrapServers(), sslBundle); + } + }; + this.contextRunner.withUserConfiguration(EnableKafkaStreamsConfiguration.class) + .withPropertyValues("spring.kafka.streams.auto-startup=false", "spring.kafka.streams.application-id=test") + .withBean(KafkaConnectionDetails.class, () -> connectionDetails) + .run((context) -> { + assertThat(context).hasSingleBean(KafkaConnectionDetails.class); + Properties configs = context + .getBean(KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME, + KafkaStreamsConfiguration.class) + .asProperties(); + assertThat(configs).containsEntry("ssl.engine.factory.class", + "org.springframework.boot.autoconfigure.kafka.SslBundleSslEngineFactory"); + assertThat(configs).containsEntry("org.springframework.boot.ssl.SslBundle", sslBundle); }); } diff --git a/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/kafka/KafkaPropertiesTests.java b/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/kafka/KafkaPropertiesTests.java index 08fb63213a..a536e0786e 100644 --- a/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/kafka/KafkaPropertiesTests.java +++ b/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/kafka/KafkaPropertiesTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2012-2024 the original author or authors. + * Copyright 2012-2025 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -107,8 +107,7 @@ class KafkaPropertiesTests { properties.getSsl().setBundle("myBundle"); Map consumerProperties = properties .buildConsumerProperties(new DefaultSslBundleRegistry("myBundle", this.sslBundle)); - assertThat(consumerProperties).containsEntry(SslConfigs.SSL_ENGINE_FACTORY_CLASS_CONFIG, - SslBundleSslEngineFactory.class.getName()); + assertThat(consumerProperties).doesNotContainKey(SslConfigs.SSL_ENGINE_FACTORY_CLASS_CONFIG); } @Test @@ -117,7 +116,7 @@ class KafkaPropertiesTests { properties.getSsl().setKeyStoreKey("-----BEGIN"); properties.getSsl().setKeyStoreLocation(new ClassPathResource("ksLoc")); assertThatExceptionOfType(MutuallyExclusiveConfigurationPropertiesException.class) - .isThrownBy(() -> properties.buildConsumerProperties()); + .isThrownBy(properties::buildConsumerProperties); } @Test @@ -126,7 +125,7 @@ class KafkaPropertiesTests { properties.getSsl().setTrustStoreLocation(new ClassPathResource("tsLoc")); properties.getSsl().setTrustStoreCertificates("-----BEGIN"); assertThatExceptionOfType(MutuallyExclusiveConfigurationPropertiesException.class) - .isThrownBy(() -> properties.buildConsumerProperties()); + .isThrownBy(properties::buildConsumerProperties); } @Test diff --git a/spring-boot-project/spring-boot-testcontainers/src/main/java/org/springframework/boot/testcontainers/service/connection/kafka/ApacheKafkaContainerConnectionDetailsFactory.java b/spring-boot-project/spring-boot-testcontainers/src/main/java/org/springframework/boot/testcontainers/service/connection/kafka/ApacheKafkaContainerConnectionDetailsFactory.java index 5815e75ac8..64deae99cd 100644 --- a/spring-boot-project/spring-boot-testcontainers/src/main/java/org/springframework/boot/testcontainers/service/connection/kafka/ApacheKafkaContainerConnectionDetailsFactory.java +++ b/spring-boot-project/spring-boot-testcontainers/src/main/java/org/springframework/boot/testcontainers/service/connection/kafka/ApacheKafkaContainerConnectionDetailsFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2012-2024 the original author or authors. + * Copyright 2012-2025 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -21,6 +21,7 @@ import java.util.List; import org.testcontainers.kafka.KafkaContainer; import org.springframework.boot.autoconfigure.kafka.KafkaConnectionDetails; +import org.springframework.boot.ssl.SslBundle; import org.springframework.boot.testcontainers.service.connection.ContainerConnectionDetailsFactory; import org.springframework.boot.testcontainers.service.connection.ContainerConnectionSource; import org.springframework.boot.testcontainers.service.connection.ServiceConnection; @@ -57,6 +58,16 @@ class ApacheKafkaContainerConnectionDetailsFactory return List.of(getContainer().getBootstrapServers()); } + @Override + public SslBundle getSslBundle() { + return super.getSslBundle(); + } + + @Override + public String getSecurityProtocol() { + return (getSslBundle() != null) ? "SSL" : "PLAINTEXT"; + } + } } diff --git a/spring-boot-project/spring-boot-testcontainers/src/main/java/org/springframework/boot/testcontainers/service/connection/kafka/ConfluentKafkaContainerConnectionDetailsFactory.java b/spring-boot-project/spring-boot-testcontainers/src/main/java/org/springframework/boot/testcontainers/service/connection/kafka/ConfluentKafkaContainerConnectionDetailsFactory.java index d7a23db295..a102b3cf50 100644 --- a/spring-boot-project/spring-boot-testcontainers/src/main/java/org/springframework/boot/testcontainers/service/connection/kafka/ConfluentKafkaContainerConnectionDetailsFactory.java +++ b/spring-boot-project/spring-boot-testcontainers/src/main/java/org/springframework/boot/testcontainers/service/connection/kafka/ConfluentKafkaContainerConnectionDetailsFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2012-2024 the original author or authors. + * Copyright 2012-2025 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -21,6 +21,7 @@ import java.util.List; import org.testcontainers.kafka.ConfluentKafkaContainer; import org.springframework.boot.autoconfigure.kafka.KafkaConnectionDetails; +import org.springframework.boot.ssl.SslBundle; import org.springframework.boot.testcontainers.service.connection.ContainerConnectionDetailsFactory; import org.springframework.boot.testcontainers.service.connection.ContainerConnectionSource; import org.springframework.boot.testcontainers.service.connection.ServiceConnection; @@ -58,6 +59,16 @@ class ConfluentKafkaContainerConnectionDetailsFactory return List.of(getContainer().getBootstrapServers()); } + @Override + public SslBundle getSslBundle() { + return super.getSslBundle(); + } + + @Override + public String getSecurityProtocol() { + return (getSslBundle() != null) ? "SSL" : "PLAINTEXT"; + } + } } diff --git a/spring-boot-project/spring-boot-testcontainers/src/main/java/org/springframework/boot/testcontainers/service/connection/kafka/DeprecatedConfluentKafkaContainerConnectionDetailsFactory.java b/spring-boot-project/spring-boot-testcontainers/src/main/java/org/springframework/boot/testcontainers/service/connection/kafka/DeprecatedConfluentKafkaContainerConnectionDetailsFactory.java index eafd40c87e..bfd862e432 100644 --- a/spring-boot-project/spring-boot-testcontainers/src/main/java/org/springframework/boot/testcontainers/service/connection/kafka/DeprecatedConfluentKafkaContainerConnectionDetailsFactory.java +++ b/spring-boot-project/spring-boot-testcontainers/src/main/java/org/springframework/boot/testcontainers/service/connection/kafka/DeprecatedConfluentKafkaContainerConnectionDetailsFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2012-2024 the original author or authors. + * Copyright 2012-2025 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -21,6 +21,7 @@ import java.util.List; import org.testcontainers.containers.KafkaContainer; import org.springframework.boot.autoconfigure.kafka.KafkaConnectionDetails; +import org.springframework.boot.ssl.SslBundle; import org.springframework.boot.testcontainers.service.connection.ContainerConnectionDetailsFactory; import org.springframework.boot.testcontainers.service.connection.ContainerConnectionSource; import org.springframework.boot.testcontainers.service.connection.ServiceConnection; @@ -59,6 +60,16 @@ class DeprecatedConfluentKafkaContainerConnectionDetailsFactory return List.of(getContainer().getBootstrapServers()); } + @Override + public SslBundle getSslBundle() { + return super.getSslBundle(); + } + + @Override + public String getSecurityProtocol() { + return (getSslBundle() != null) ? "SSL" : "PLAINTEXT"; + } + } } diff --git a/spring-boot-project/spring-boot-testcontainers/src/main/java/org/springframework/boot/testcontainers/service/connection/redpanda/RedpandaContainerConnectionDetailsFactory.java b/spring-boot-project/spring-boot-testcontainers/src/main/java/org/springframework/boot/testcontainers/service/connection/redpanda/RedpandaContainerConnectionDetailsFactory.java index 48ec53bec1..86693d18f0 100644 --- a/spring-boot-project/spring-boot-testcontainers/src/main/java/org/springframework/boot/testcontainers/service/connection/redpanda/RedpandaContainerConnectionDetailsFactory.java +++ b/spring-boot-project/spring-boot-testcontainers/src/main/java/org/springframework/boot/testcontainers/service/connection/redpanda/RedpandaContainerConnectionDetailsFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2012-2023 the original author or authors. + * Copyright 2012-2025 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -21,6 +21,7 @@ import java.util.List; import org.testcontainers.redpanda.RedpandaContainer; import org.springframework.boot.autoconfigure.kafka.KafkaConnectionDetails; +import org.springframework.boot.ssl.SslBundle; import org.springframework.boot.testcontainers.service.connection.ContainerConnectionDetailsFactory; import org.springframework.boot.testcontainers.service.connection.ContainerConnectionSource; import org.springframework.boot.testcontainers.service.connection.ServiceConnection; @@ -55,6 +56,16 @@ class RedpandaContainerConnectionDetailsFactory return List.of(getContainer().getBootstrapServers()); } + @Override + public SslBundle getSslBundle() { + return super.getSslBundle(); + } + + @Override + public String getSecurityProtocol() { + return (getSslBundle() != null) ? "SSL" : "PLAINTEXT"; + } + } } diff --git a/spring-boot-tests/spring-boot-smoke-tests/spring-boot-smoke-test-kafka/build.gradle b/spring-boot-tests/spring-boot-smoke-tests/spring-boot-smoke-test-kafka/build.gradle index 8e955e4afb..23f257e59d 100644 --- a/spring-boot-tests/spring-boot-smoke-tests/spring-boot-smoke-test-kafka/build.gradle +++ b/spring-boot-tests/spring-boot-smoke-tests/spring-boot-smoke-test-kafka/build.gradle @@ -18,6 +18,7 @@ configurations.all { dependencies { dockerTestImplementation(project(":spring-boot-project:spring-boot-starters:spring-boot-starter-test")) dockerTestImplementation(project(":spring-boot-project:spring-boot-tools:spring-boot-test-support-docker")) + dockerTestImplementation(project(":spring-boot-project:spring-boot-testcontainers")) dockerTestImplementation("org.awaitility:awaitility") dockerTestImplementation("org.testcontainers:junit-jupiter") dockerTestImplementation("org.testcontainers:kafka") diff --git a/spring-boot-tests/spring-boot-smoke-tests/spring-boot-smoke-test-kafka/src/dockerTest/java/smoketest/kafka/ssl/SampleKafkaSslApplicationTests.java b/spring-boot-tests/spring-boot-smoke-tests/spring-boot-smoke-test-kafka/src/dockerTest/java/smoketest/kafka/ssl/SampleKafkaSslApplicationTests.java index ddd17bfc05..6c4556e1af 100644 --- a/spring-boot-tests/spring-boot-smoke-tests/spring-boot-smoke-test-kafka/src/dockerTest/java/smoketest/kafka/ssl/SampleKafkaSslApplicationTests.java +++ b/spring-boot-tests/spring-boot-smoke-tests/spring-boot-smoke-test-kafka/src/dockerTest/java/smoketest/kafka/ssl/SampleKafkaSslApplicationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2012-2024 the original author or authors. + * Copyright 2012-2025 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -23,16 +23,16 @@ import org.junit.jupiter.api.Test; import org.testcontainers.junit.jupiter.Container; import org.testcontainers.junit.jupiter.Testcontainers; import org.testcontainers.kafka.ConfluentKafkaContainer; -import org.testcontainers.utility.MountableFile; import smoketest.kafka.Consumer; import smoketest.kafka.Producer; import smoketest.kafka.SampleMessage; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.testcontainers.service.connection.JksKeyStore; +import org.springframework.boot.testcontainers.service.connection.JksTrustStore; +import org.springframework.boot.testcontainers.service.connection.ServiceConnection; import org.springframework.boot.testsupport.container.TestImage; -import org.springframework.test.context.DynamicPropertyRegistry; -import org.springframework.test.context.DynamicPropertySource; import static org.assertj.core.api.Assertions.assertThat; import static org.hamcrest.Matchers.empty; @@ -45,38 +45,14 @@ import static org.hamcrest.Matchers.not; * @author Eddú Meléndez */ @Testcontainers(disabledWithoutDocker = true) -@SpringBootTest(classes = { SampleKafkaSslApplication.class, Producer.class, Consumer.class }, - properties = { "spring.kafka.security.protocol=SSL", - "spring.kafka.properties.ssl.endpoint.identification.algorithm=", "spring.kafka.ssl.bundle=client", - "spring.ssl.bundle.jks.client.keystore.location=classpath:ssl/test-client.p12", - "spring.ssl.bundle.jks.client.keystore.password=password", - "spring.ssl.bundle.jks.client.truststore.location=classpath:ssl/test-ca.p12", - "spring.ssl.bundle.jks.client.truststore.password=password" }) +@SpringBootTest(classes = { SampleKafkaSslApplication.class, Producer.class, Consumer.class }) class SampleKafkaSslApplicationTests { @Container - public static ConfluentKafkaContainer kafka = TestImage.container(ConfluentKafkaContainer.class) - .withEnv("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP", "PLAINTEXT:SSL,BROKER:PLAINTEXT,CONTROLLER:PLAINTEXT") - .withEnv("KAFKA_AUTO_CREATE_TOPICS_ENABLE", "true") - .withEnv("KAFKA_SSL_CLIENT_AUTH", "required") - .withEnv("KAFKA_SSL_KEYSTORE_LOCATION", "/etc/kafka/secrets/certs/test-server.p12") - .withEnv("KAFKA_SSL_KEYSTORE_PASSWORD", "password") - .withEnv("KAFKA_SSL_KEY_PASSWORD", "password") - .withEnv("KAFKA_SSL_TRUSTSTORE_LOCATION", "/etc/kafka/secrets/certs/test-ca.p12") - .withEnv("KAFKA_SSL_TRUSTSTORE_PASSWORD", "password") - .withEnv("KAFKA_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM", "") - .withCopyFileToContainer(MountableFile.forClasspathResource("ssl/test-server.p12"), - "/etc/kafka/secrets/certs/test-server.p12") - .withCopyFileToContainer(MountableFile.forClasspathResource("ssl/credentials"), - "/etc/kafka/secrets/certs/credentials") - .withCopyFileToContainer(MountableFile.forClasspathResource("ssl/test-ca.p12"), - "/etc/kafka/secrets/certs/test-ca.p12"); - - @DynamicPropertySource - static void kafkaProperties(DynamicPropertyRegistry registry) { - registry.add("spring.kafka.bootstrap-servers", - () -> String.format("%s:%s", kafka.getHost(), kafka.getMappedPort(9092))); - } + @ServiceConnection + @JksTrustStore(location = "classpath:ssl/test-ca.p12", password = "password") + @JksKeyStore(location = "classpath:ssl/test-client.p12", password = "password") + public static ConfluentKafkaContainer kafka = TestImage.container(SecureKafkaContainer.class); @Autowired private Producer producer; diff --git a/spring-boot-tests/spring-boot-smoke-tests/spring-boot-smoke-test-kafka/src/dockerTest/java/smoketest/kafka/ssl/SecureKafkaContainer.java b/spring-boot-tests/spring-boot-smoke-tests/spring-boot-smoke-test-kafka/src/dockerTest/java/smoketest/kafka/ssl/SecureKafkaContainer.java new file mode 100644 index 0000000000..908217afe9 --- /dev/null +++ b/spring-boot-tests/spring-boot-smoke-tests/spring-boot-smoke-test-kafka/src/dockerTest/java/smoketest/kafka/ssl/SecureKafkaContainer.java @@ -0,0 +1,56 @@ +/* + * Copyright 2012-2025 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package smoketest.kafka.ssl; + +import org.testcontainers.kafka.ConfluentKafkaContainer; +import org.testcontainers.utility.DockerImageName; +import org.testcontainers.utility.MountableFile; + +/** + * Kafka container with SSL enabled. + * + * @author Scott Frederick + * @author Eddú Meléndez + * @author Moritz Halbritter + */ +class SecureKafkaContainer extends ConfluentKafkaContainer { + + SecureKafkaContainer(DockerImageName dockerImageName) { + super(dockerImageName); + } + + @Override + protected void configure() { + super.configure(); + withEnv("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP", "PLAINTEXT:SSL,BROKER:PLAINTEXT,CONTROLLER:PLAINTEXT") + .withEnv("KAFKA_AUTO_CREATE_TOPICS_ENABLE", "true") + .withEnv("KAFKA_SSL_CLIENT_AUTH", "required") + .withEnv("KAFKA_SSL_KEYSTORE_LOCATION", "/etc/kafka/secrets/certs/test-server.p12") + .withEnv("KAFKA_SSL_KEYSTORE_PASSWORD", "password") + .withEnv("KAFKA_SSL_KEY_PASSWORD", "password") + .withEnv("KAFKA_SSL_TRUSTSTORE_LOCATION", "/etc/kafka/secrets/certs/test-ca.p12") + .withEnv("KAFKA_SSL_TRUSTSTORE_PASSWORD", "password") + .withEnv("KAFKA_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM", ""); + withCopyFileToContainer(MountableFile.forClasspathResource("ssl/test-server.p12"), + "/etc/kafka/secrets/certs/test-server.p12"); + withCopyFileToContainer(MountableFile.forClasspathResource("ssl/credentials"), + "/etc/kafka/secrets/certs/credentials"); + withCopyFileToContainer(MountableFile.forClasspathResource("ssl/test-ca.p12"), + "/etc/kafka/secrets/certs/test-ca.p12"); + } + +} diff --git a/spring-boot-tests/spring-boot-smoke-tests/spring-boot-smoke-test-kafka/src/test/resources/ssl/credentials b/spring-boot-tests/spring-boot-smoke-tests/spring-boot-smoke-test-kafka/src/dockerTest/resources/ssl/credentials similarity index 100% rename from spring-boot-tests/spring-boot-smoke-tests/spring-boot-smoke-test-kafka/src/test/resources/ssl/credentials rename to spring-boot-tests/spring-boot-smoke-tests/spring-boot-smoke-test-kafka/src/dockerTest/resources/ssl/credentials diff --git a/spring-boot-tests/spring-boot-smoke-tests/spring-boot-smoke-test-kafka/src/test/resources/ssl/test-ca.p12 b/spring-boot-tests/spring-boot-smoke-tests/spring-boot-smoke-test-kafka/src/dockerTest/resources/ssl/test-ca.p12 similarity index 100% rename from spring-boot-tests/spring-boot-smoke-tests/spring-boot-smoke-test-kafka/src/test/resources/ssl/test-ca.p12 rename to spring-boot-tests/spring-boot-smoke-tests/spring-boot-smoke-test-kafka/src/dockerTest/resources/ssl/test-ca.p12 diff --git a/spring-boot-tests/spring-boot-smoke-tests/spring-boot-smoke-test-kafka/src/test/resources/ssl/test-client.p12 b/spring-boot-tests/spring-boot-smoke-tests/spring-boot-smoke-test-kafka/src/dockerTest/resources/ssl/test-client.p12 similarity index 100% rename from spring-boot-tests/spring-boot-smoke-tests/spring-boot-smoke-test-kafka/src/test/resources/ssl/test-client.p12 rename to spring-boot-tests/spring-boot-smoke-tests/spring-boot-smoke-test-kafka/src/dockerTest/resources/ssl/test-client.p12 diff --git a/spring-boot-tests/spring-boot-smoke-tests/spring-boot-smoke-test-kafka/src/test/resources/ssl/test-server.p12 b/spring-boot-tests/spring-boot-smoke-tests/spring-boot-smoke-test-kafka/src/dockerTest/resources/ssl/test-server.p12 similarity index 100% rename from spring-boot-tests/spring-boot-smoke-tests/spring-boot-smoke-test-kafka/src/test/resources/ssl/test-server.p12 rename to spring-boot-tests/spring-boot-smoke-tests/spring-boot-smoke-test-kafka/src/dockerTest/resources/ssl/test-server.p12