spring/spring-boot
1
0
Fork 0
mirror of https://github.com/spring-projects/spring-boot.git synced 2026-09-29 21:59:02 +00:00
Code Issues Projects Releases Wiki Activity

Apply consumer-specific security protocol

Browse Source 
Closes gh-51365
This commit is contained in:
Stéphane Nicoll
2026-08-11 17:46:39 +02:00
parent 99ba99d3b3
commit ada7ac7c48
2 changed files with 32 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
module/spring-boot-kafka/src/main/java/org/springframework/boot/kafka/autoconfigure/KafkaAutoConfiguration.java
+1 -1
View File
@@ -206,7 +206,7 @@ public final class KafkaAutoConfiguration {
KafkaConnectionDetails connectionDetails) {
Configuration consumer = connectionDetails.getConsumer();
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, consumer.getBootstrapServers());
applySecurityProtocol(properties, connectionDetails.getSecurityProtocol());
applySecurityProtocol(properties, consumer.getSecurityProtocol());
applySslBundle(properties, consumer.getSslBundle());
}
module/spring-boot-kafka/src/test/java/org/springframework/boot/kafka/autoconfigure/KafkaAutoConfigurationTests.java
+31 -1
View File
@@ -993,7 +993,37 @@ class KafkaAutoConfigurationTests {
}
@Test
void specificSecurityProtocolOverridesCommonSecurityProtocol() {
void specificProducerSecurityProtocolOverridesCommonSecurityProtocol() {
this.contextRunner
.withPropertyValues("spring.kafka.security.protocol=SSL",
"spring.kafka.producer.security.protocol=PLAINTEXT")
.run((context) -> {
DefaultKafkaConsumerFactory<?, ?> consumerFactory = context.getBean(DefaultKafkaConsumerFactory.class);
Map<String, Object> consumerConfigs = consumerFactory.getConfigurationProperties();
assertThat(consumerConfigs).containsEntry(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");
DefaultKafkaProducerFactory<?, ?> producerFactory = context.getBean(DefaultKafkaProducerFactory.class);
Map<String, Object> producerConfigs = producerFactory.getConfigurationProperties();
assertThat(producerConfigs).containsEntry(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "PLAINTEXT");
});
}
@Test
void specificConsumerSecurityProtocolOverridesCommonSecurityProtocol() {
this.contextRunner
.withPropertyValues("spring.kafka.security.protocol=SSL",
"spring.kafka.consumer.security.protocol=PLAINTEXT")
.run((context) -> {
DefaultKafkaProducerFactory<?, ?> producerFactory = context.getBean(DefaultKafkaProducerFactory.class);
Map<String, Object> producerConfigs = producerFactory.getConfigurationProperties();
assertThat(producerConfigs).containsEntry(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");
DefaultKafkaConsumerFactory<?, ?> consumerFactory = context.getBean(DefaultKafkaConsumerFactory.class);
Map<String, Object> consumerConfigs = consumerFactory.getConfigurationProperties();
assertThat(consumerConfigs).containsEntry(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "PLAINTEXT");
});
}
@Test
void specificAdminSecurityProtocolOverridesCommonSecurityProtocol() {
this.contextRunner
.withPropertyValues("spring.kafka.security.protocol=SSL", "spring.kafka.admin.security.protocol=PLAINTEXT")
.run((context) -> {