Skip to content

Commit dae891f

Browse files
mhalbrittermhalbritter
committed
Add SSL service connection support for Kafka
See gh-41137
1 parent 789d30d commit dae891f

File tree

19 files changed

+477
-111
lines changed

19 files changed

+477
-111
lines changed

‎spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaAnnotationDrivenConfiguration.java

Lines changed: 4 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2012-2023 the original author or authors.
2+
* Copyright 2012-2025 the original author or authors.
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -23,7 +23,6 @@
2323
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
2424
import org.springframework.boot.autoconfigure.condition.ConditionalOnThreading;
2525
import org.springframework.boot.autoconfigure.thread.Threading;
26-
import org.springframework.boot.ssl.SslBundles;
2726
import org.springframework.context.annotation.Bean;
2827
import org.springframework.context.annotation.Configuration;
2928
import org.springframework.core.task.SimpleAsyncTaskExecutor;
@@ -152,11 +151,10 @@ private ConcurrentKafkaListenerContainerFactoryConfigurer configurer() {
152151
ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
153152
ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
154153
ObjectProvider<ConsumerFactory<Object, Object>> kafkaConsumerFactory,
155-
ObjectProvider<ContainerCustomizer<Object, Object, ConcurrentMessageListenerContainer<Object, Object>>> kafkaContainerCustomizer,
156-
ObjectProvider<SslBundles> sslBundles) {
154+
ObjectProvider<ContainerCustomizer<Object, Object, ConcurrentMessageListenerContainer<Object, Object>>> kafkaContainerCustomizer) {
157155
ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
158-
configurer.configure(factory, kafkaConsumerFactory.getIfAvailable(() -> new DefaultKafkaConsumerFactory<>(
159-
this.properties.buildConsumerProperties(sslBundles.getIfAvailable()))));
156+
configurer.configure(factory, kafkaConsumerFactory
157+
.getIfAvailable(() -> new DefaultKafkaConsumerFactory<>(this.properties.buildConsumerProperties())));
160158
kafkaContainerCustomizer.ifAvailable(factory::setContainerCustomizer);
161159
return factory;
162160
}

‎spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfiguration.java

Lines changed: 40 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
import org.apache.kafka.clients.CommonClientConfigs;
2424
import org.apache.kafka.clients.consumer.ConsumerConfig;
2525
import org.apache.kafka.clients.producer.ProducerConfig;
26+
import org.apache.kafka.common.config.SslConfigs;
2627

2728
import org.springframework.beans.factory.ObjectProvider;
2829
import org.springframework.boot.autoconfigure.AutoConfiguration;
@@ -32,10 +33,12 @@
3233
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
3334
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
3435
import org.springframework.boot.autoconfigure.condition.ConditionalOnSingleCandidate;
36+
import org.springframework.boot.autoconfigure.kafka.KafkaConnectionDetails.Configuration;
3537
import org.springframework.boot.autoconfigure.kafka.KafkaProperties.Jaas;
3638
import org.springframework.boot.autoconfigure.kafka.KafkaProperties.Retry.Topic.Backoff;
3739
import org.springframework.boot.context.properties.EnableConfigurationProperties;
3840
import org.springframework.boot.context.properties.PropertyMapper;
41+
import org.springframework.boot.ssl.SslBundle;
3942
import org.springframework.boot.ssl.SslBundles;
4043
import org.springframework.context.annotation.Bean;
4144
import org.springframework.context.annotation.Import;
@@ -54,6 +57,7 @@
5457
import org.springframework.kafka.transaction.KafkaTransactionManager;
5558
import org.springframework.retry.backoff.BackOffPolicyBuilder;
5659
import org.springframework.retry.backoff.SleepingBackOffPolicy;
60+
import org.springframework.util.StringUtils;
5761

5862
/**
5963
* {@link EnableAutoConfiguration Auto-configuration} for Apache Kafka.
@@ -84,8 +88,9 @@ public class KafkaAutoConfiguration {
8488

8589
@Bean
8690
@ConditionalOnMissingBean(KafkaConnectionDetails.class)
87-
PropertiesKafkaConnectionDetails kafkaConnectionDetails(KafkaProperties properties) {
88-
return new PropertiesKafkaConnectionDetails(properties);
91+
PropertiesKafkaConnectionDetails kafkaConnectionDetails(KafkaProperties properties,
92+
ObjectProvider<SslBundles> sslBundles) {
93+
return new PropertiesKafkaConnectionDetails(properties, sslBundles.getIfAvailable());
8994
}
9095

9196
@Bean
@@ -111,9 +116,9 @@ public LoggingProducerListener<Object, Object> kafkaProducerListener() {
111116

112117
@Bean
113118
@ConditionalOnMissingBean(ConsumerFactory.class)
114-
public DefaultKafkaConsumerFactory<?, ?> kafkaConsumerFactory(KafkaConnectionDetails connectionDetails,
115-
ObjectProvider<DefaultKafkaConsumerFactoryCustomizer> customizers, ObjectProvider<SslBundles> sslBundles) {
116-
Map<String, Object> properties = this.properties.buildConsumerProperties(sslBundles.getIfAvailable());
119+
DefaultKafkaConsumerFactory<?, ?> kafkaConsumerFactory(KafkaConnectionDetails connectionDetails,
120+
ObjectProvider<DefaultKafkaConsumerFactoryCustomizer> customizers) {
121+
Map<String, Object> properties = this.properties.buildConsumerProperties();
117122
applyKafkaConnectionDetailsForConsumer(properties, connectionDetails);
118123
DefaultKafkaConsumerFactory<Object, Object> factory = new DefaultKafkaConsumerFactory<>(properties);
119124
customizers.orderedStream().forEach((customizer) -> customizer.customize(factory));
@@ -122,9 +127,9 @@ public LoggingProducerListener<Object, Object> kafkaProducerListener() {
122127

123128
@Bean
124129
@ConditionalOnMissingBean(ProducerFactory.class)
125-
public DefaultKafkaProducerFactory<?, ?> kafkaProducerFactory(KafkaConnectionDetails connectionDetails,
126-
ObjectProvider<DefaultKafkaProducerFactoryCustomizer> customizers, ObjectProvider<SslBundles> sslBundles) {
127-
Map<String, Object> properties = this.properties.buildProducerProperties(sslBundles.getIfAvailable());
130+
DefaultKafkaProducerFactory<?, ?> kafkaProducerFactory(KafkaConnectionDetails connectionDetails,
131+
ObjectProvider<DefaultKafkaProducerFactoryCustomizer> customizers) {
132+
Map<String, Object> properties = this.properties.buildProducerProperties();
128133
applyKafkaConnectionDetailsForProducer(properties, connectionDetails);
129134
DefaultKafkaProducerFactory<?, ?> factory = new DefaultKafkaProducerFactory<>(properties);
130135
String transactionIdPrefix = this.properties.getProducer().getTransactionIdPrefix();
@@ -160,8 +165,8 @@ public KafkaJaasLoginModuleInitializer kafkaJaasInitializer() throws IOException
160165

161166
@Bean
162167
@ConditionalOnMissingBean
163-
public KafkaAdmin kafkaAdmin(KafkaConnectionDetails connectionDetails, ObjectProvider<SslBundles> sslBundles) {
164-
Map<String, Object> properties = this.properties.buildAdminProperties(sslBundles.getIfAvailable());
168+
KafkaAdmin kafkaAdmin(KafkaConnectionDetails connectionDetails) {
169+
Map<String, Object> properties = this.properties.buildAdminProperties(null);
165170
applyKafkaConnectionDetailsForAdmin(properties, connectionDetails);
166171
KafkaAdmin kafkaAdmin = new KafkaAdmin(properties);
167172
KafkaProperties.Admin admin = this.properties.getAdmin();
@@ -193,26 +198,26 @@ public RetryTopicConfiguration kafkaRetryTopicConfiguration(KafkaTemplate<?, ?>
193198

194199
private void applyKafkaConnectionDetailsForConsumer(Map<String, Object> properties,
195200
KafkaConnectionDetails connectionDetails) {
196-
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, connectionDetails.getConsumerBootstrapServers());
197-
if (!(connectionDetails instanceof PropertiesKafkaConnectionDetails)) {
198-
properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "PLAINTEXT");
199-
}
201+
Configuration consumer = connectionDetails.getConsumer();
202+
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, consumer.getBootstrapServers());
203+
applySecurityProtocol(properties, connectionDetails.getSecurityProtocol());
204+
applySslBundle(properties, consumer.getSslBundle());
200205
}
201206

202207
private void applyKafkaConnectionDetailsForProducer(Map<String, Object> properties,
203208
KafkaConnectionDetails connectionDetails) {
204-
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, connectionDetails.getProducerBootstrapServers());
205-
if (!(connectionDetails instanceof PropertiesKafkaConnectionDetails)) {
206-
properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "PLAINTEXT");
207-
}
209+
Configuration producer = connectionDetails.getProducer();
210+
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, producer.getBootstrapServers());
211+
applySecurityProtocol(properties, producer.getSecurityProtocol());
212+
applySslBundle(properties, producer.getSslBundle());
208213
}
209214

210215
private void applyKafkaConnectionDetailsForAdmin(Map<String, Object> properties,
211216
KafkaConnectionDetails connectionDetails) {
212-
properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, connectionDetails.getAdminBootstrapServers());
213-
if (!(connectionDetails instanceof PropertiesKafkaConnectionDetails)) {
214-
properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "PLAINTEXT");
215-
}
217+
Configuration admin = connectionDetails.getAdmin();
218+
properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, admin.getBootstrapServers());
219+
applySecurityProtocol(properties, admin.getSecurityProtocol());
220+
applySslBundle(properties, admin.getSslBundle());
216221
}
217222

218223
private static void setBackOffPolicy(RetryTopicConfigurationBuilder builder, Backoff retryTopicBackoff) {
@@ -231,4 +236,17 @@ private static void setBackOffPolicy(RetryTopicConfigurationBuilder builder, Bac
231236
}
232237
}
233238

239+
static void applySslBundle(Map<String, Object> properties, SslBundle sslBundle) {
240+
if (sslBundle != null) {
241+
properties.put(SslConfigs.SSL_ENGINE_FACTORY_CLASS_CONFIG, SslBundleSslEngineFactory.class.getName());
242+
properties.put(SslBundle.class.getName(), sslBundle);
243+
}
244+
}
245+
246+
static void applySecurityProtocol(Map<String, Object> properties, String securityProtocol) {
247+
if (StringUtils.hasLength(securityProtocol)) {
248+
properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, securityProtocol);
249+
}
250+
}
251+
234252
}

‎spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaConnectionDetails.java

Lines changed: 143 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2012-2023 the original author or authors.
2+
* Copyright 2012-2025 the original author or authors.
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -19,6 +19,7 @@
1919
import java.util.List;
2020

2121
import org.springframework.boot.autoconfigure.service.connection.ConnectionDetails;
22+
import org.springframework.boot.ssl.SslBundle;
2223

2324
/**
2425
* Details required to establish a connection to a Kafka service.
@@ -36,36 +37,173 @@ public interface KafkaConnectionDetails extends ConnectionDetails {
3637
*/
3738
List<String> getBootstrapServers();
3839

40+
/**
41+
* Returns the SSL bundle.
42+
* @return the SSL bundle
43+
* @since 3.5.0
44+
*/
45+
default SslBundle getSslBundle() {
46+
return null;
47+
}
48+
49+
/**
50+
* Returns the security protocol.
51+
* @return the security protocol
52+
* @since 3.5.0
53+
*/
54+
default String getSecurityProtocol() {
55+
return null;
56+
}
57+
58+
/**
59+
* Returns the consumer configuration.
60+
* @return the consumer configuration
61+
* @since 3.5.0
62+
*/
63+
default Configuration getConsumer() {
64+
return Configuration.of(getBootstrapServers(), getSslBundle(), getSecurityProtocol());
65+
}
66+
67+
/**
68+
* Returns the producer configuration.
69+
* @return the producer configuration
70+
* @since 3.5.0
71+
*/
72+
default Configuration getProducer() {
73+
return Configuration.of(getBootstrapServers(), getSslBundle(), getSecurityProtocol());
74+
}
75+
76+
/**
77+
* Returns the admin configuration.
78+
* @return the admin configuration
79+
* @since 3.5.0
80+
*/
81+
default Configuration getAdmin() {
82+
return Configuration.of(getBootstrapServers(), getSslBundle(), getSecurityProtocol());
83+
}
84+
85+
/**
86+
* Returns the Kafka Streams configuration.
87+
* @return the Kafka Streams configuration
88+
* @since 3.5.0
89+
*/
90+
default Configuration getStreams() {
91+
return Configuration.of(getBootstrapServers(), getSslBundle(), getSecurityProtocol());
92+
}
93+
3994
/**
4095
* Returns the list of bootstrap servers used for consumers.
4196
* @return the list of bootstrap servers used for consumers
97+
* @deprecated since 3.5.0 for removal in 3.7.0 in favor of {@link #getConsumer()}
4298
*/
99+
@Deprecated(since = "3.5.0", forRemoval = true)
43100
default List<String> getConsumerBootstrapServers() {
44-
return getBootstrapServers();
101+
return getConsumer().getBootstrapServers();
45102
}
46103

47104
/**
48105
* Returns the list of bootstrap servers used for producers.
49106
* @return the list of bootstrap servers used for producers
107+
* @deprecated since 3.5.0 for removal in 3.7.0 in favor of {@link #getProducer()}
50108
*/
109+
@Deprecated(since = "3.5.0", forRemoval = true)
51110
default List<String> getProducerBootstrapServers() {
52-
return getBootstrapServers();
111+
return getProducer().getBootstrapServers();
53112
}
54113

55114
/**
56115
* Returns the list of bootstrap servers used for the admin.
57116
* @return the list of bootstrap servers used for the admin
117+
* @deprecated since 3.5.0 for removal in 3.7.0 in favor of {@link #getAdmin()}
58118
*/
119+
@Deprecated(since = "3.5.0", forRemoval = true)
59120
default List<String> getAdminBootstrapServers() {
60-
return getBootstrapServers();
121+
return getAdmin().getBootstrapServers();
61122
}
62123

63124
/**
64125
* Returns the list of bootstrap servers used for Kafka Streams.
65126
* @return the list of bootstrap servers used for Kafka Streams
127+
* @deprecated since 3.5.0 for removal in 3.7.0 in favor of {@link #getStreams()}
66128
*/
129+
@Deprecated(since = "3.5.0", forRemoval = true)
67130
default List<String> getStreamsBootstrapServers() {
68-
return getBootstrapServers();
131+
return getStreams().getBootstrapServers();
132+
}
133+
134+
/**
135+
* Kafka connection details configuration.
136+
*/
137+
interface Configuration {
138+
139+
/**
140+
* Creates a new configuration with the given bootstrap servers.
141+
* @param bootstrapServers the bootstrap servers
142+
* @return the configuration
143+
*/
144+
static Configuration of(List<String> bootstrapServers) {
145+
return Configuration.of(bootstrapServers, null, null);
146+
}
147+
148+
/**
149+
* Creates a new configuration with the given bootstrap servers and SSL bundle.
150+
* @param bootstrapServers the bootstrap servers
151+
* @param sslBundle the SSL bundle
152+
* @return the configuration
153+
*/
154+
static Configuration of(List<String> bootstrapServers, SslBundle sslBundle) {
155+
return Configuration.of(bootstrapServers, sslBundle, null);
156+
}
157+
158+
/**
159+
* Creates a new configuration with the given bootstrap servers, SSL bundle and
160+
* security protocol.
161+
* @param bootstrapServers the bootstrap servers
162+
* @param sslBundle the SSL bundle
163+
* @param securityProtocol the security protocol
164+
* @return the configuration
165+
*/
166+
static Configuration of(List<String> bootstrapServers, SslBundle sslBundle, String securityProtocol) {
167+
return new Configuration() {
168+
@Override
169+
public List<String> getBootstrapServers() {
170+
return bootstrapServers;
171+
}
172+
173+
@Override
174+
public SslBundle getSslBundle() {
175+
return sslBundle;
176+
}
177+
178+
@Override
179+
public String getSecurityProtocol() {
180+
return securityProtocol;
181+
}
182+
};
183+
}
184+
185+
/**
186+
* Returns the list of bootstrap servers.
187+
* @return the list of bootstrap servers
188+
*/
189+
List<String> getBootstrapServers();
190+
191+
/**
192+
* Returns the SSL bundle.
193+
* @return the SSL bundle
194+
*/
195+
default SslBundle getSslBundle() {
196+
return null;
197+
}
198+
199+
/**
200+
* Returns the security protocol.
201+
* @return the security protocol
202+
*/
203+
default String getSecurityProtocol() {
204+
return null;
205+
}
206+
69207
}
70208

71209
}

‎spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaProperties.java

Lines changed: 2 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,6 @@
3838
import org.springframework.boot.context.properties.PropertyMapper;
3939
import org.springframework.boot.context.properties.source.MutuallyExclusiveConfigurationPropertiesException;
4040
import org.springframework.boot.convert.DurationUnit;
41-
import org.springframework.boot.ssl.SslBundle;
4241
import org.springframework.boot.ssl.SslBundles;
4342
import org.springframework.core.io.Resource;
4443
import org.springframework.kafka.listener.ContainerProperties.AckMode;
@@ -1401,10 +1400,10 @@ public Map<String, Object> buildProperties() {
14011400
public Map<String, Object> buildProperties(SslBundles sslBundles) {
14021401
validate();
14031402
String bundleName = getBundle();
1403+
Properties properties = new Properties();
14041404
if (StringUtils.hasText(bundleName)) {
1405-
return buildPropertiesForSslBundle(sslBundles, bundleName);
1405+
return properties;
14061406
}
1407-
Properties properties = new Properties();
14081407
PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
14091408
map.from(this::getKeyPassword).to(properties.in(SslConfigs.SSL_KEY_PASSWORD_CONFIG));
14101409
map.from(this::getKeyStoreCertificateChain)
@@ -1425,13 +1424,6 @@ public Map<String, Object> buildProperties(SslBundles sslBundles) {
14251424
return properties;
14261425
}
14271426

1428-
private Map<String, Object> buildPropertiesForSslBundle(SslBundles sslBundles, String name) {
1429-
Properties properties = new Properties();
1430-
properties.in(SslConfigs.SSL_ENGINE_FACTORY_CLASS_CONFIG).accept(SslBundleSslEngineFactory.class.getName());
1431-
properties.in(SslBundle.class.getName()).accept(sslBundles.getBundle(name));
1432-
return properties;
1433-
}
1434-
14351427
private void validate() {
14361428
MutuallyExclusiveConfigurationPropertiesException.throwIfMultipleMatchingValuesIn((entries) -> {
14371429
entries.put("spring.kafka.ssl.key-store-key", getKeyStoreKey());

0 commit comments

Comments
 (0)