SpringCloud/spring-cloud-task
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Kafka requires SslBundles for consumer and producer properties

Browse Source
This commit is contained in:
Glenn Renfro
2023-10-19 11:52:44 -04:00
parent b06ca44ba0
commit f8d3c66c1a
4 changed files with 51 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
spring-cloud-starter-single-step-batch-job/src/main/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemReaderAutoConfiguration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2022 the original author or authors.
* Copyright 2020-2023 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.
@@ -31,6 +31,7 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.boot.ssl.SslBundles;
import org.springframework.context.annotation.Bean;
import org.springframework.util.StringUtils;
@@ -50,13 +51,16 @@ public class KafkaItemReaderAutoConfiguration {
@Autowired
private KafkaProperties kafkaProperties;
@Autowired
private SslBundles sslBundles;
@Bean
@ConditionalOnMissingBean
@ConditionalOnProperty(prefix = "spring.batch.job.kafkaitemreader", name = "name")
public KafkaItemReader<Object, Map<String, Object>> kafkaItemReader(
KafkaItemReaderProperties kafkaItemReaderProperties) {
Properties consumerProperties = new Properties();
consumerProperties.putAll(this.kafkaProperties.getConsumer().buildProperties());
consumerProperties.putAll(this.kafkaProperties.getConsumer().buildProperties(sslBundles));
validateProperties(kafkaItemReaderProperties);
if (kafkaItemReaderProperties.getPartitions() == null
|| kafkaItemReaderProperties.getPartitions().size() == 0) {

7
spring-cloud-starter-single-step-batch-job/src/main/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemWriterAutoConfiguration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2022 the original author or authors.
* Copyright 2020-2023 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.
@@ -30,6 +30,7 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.boot.ssl.SslBundles;
import org.springframework.context.annotation.Bean;
import org.springframework.core.convert.converter.Converter;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
@@ -84,9 +85,9 @@ public class KafkaItemWriterAutoConfiguration {
@Bean
@ConditionalOnMissingBean
ProducerFactory<Object, Map<String, Object>> producerFactory() {
ProducerFactory<Object, Map<String, Object>> producerFactory(SslBundles sslBundles) {
Map<String, Object> configs = new HashMap<>();
configs.putAll(this.kafkaProperties.getProducer().buildProperties());
configs.putAll(this.kafkaProperties.getProducer().buildProperties(sslBundles));
return new DefaultKafkaProducerFactory<>(configs, null, new JsonSerializer<>());
}

22
spring-cloud-starter-single-step-batch-job/src/test/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemReaderAutoConfigurationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2022 the original author or authors.
* Copyright 2020-2023 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.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.producer.Producer;
@@ -40,6 +41,9 @@ import org.springframework.boot.autoconfigure.AutoConfigurations;
import org.springframework.boot.autoconfigure.batch.BatchAutoConfiguration;
import org.springframework.boot.autoconfigure.context.PropertyPlaceholderAutoConfiguration;
import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
import org.springframework.boot.ssl.NoSuchSslBundleException;
import org.springframework.boot.ssl.SslBundle;
import org.springframework.boot.ssl.SslBundles;
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
import org.springframework.cloud.task.batch.autoconfigure.SingleStepJobAutoConfiguration;
import org.springframework.context.annotation.Bean;
@@ -227,6 +231,22 @@ public class KafkaItemReaderAutoConfigurationTests {
return new ListItemWriter<>();
}
@Bean
public SslBundles sslBundles() {
return new SslBundles() {
@Override
public SslBundle getBundle(String name) throws NoSuchSslBundleException {
return null;
}
@Override
public void addBundleUpdateHandler(String name, Consumer<SslBundle> updateHandler)
throws NoSuchSslBundleException {
}
};
}
}
}

21
spring-cloud-starter-single-step-batch-job/src/test/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemWriterTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2022 the original author or authors.
* Copyright 2020-2023 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.
@@ -38,6 +38,9 @@ import org.springframework.boot.autoconfigure.AutoConfigurations;
import org.springframework.boot.autoconfigure.batch.BatchAutoConfiguration;
import org.springframework.boot.autoconfigure.context.PropertyPlaceholderAutoConfiguration;
import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
import org.springframework.boot.ssl.NoSuchSslBundleException;
import org.springframework.boot.ssl.SslBundle;
import org.springframework.boot.ssl.SslBundles;
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
import org.springframework.cloud.task.batch.autoconfigure.SingleStepJobAutoConfiguration;
import org.springframework.context.ApplicationContext;
@@ -143,6 +146,22 @@ public class KafkaItemWriterTests {
itemReaderList.add(prepMap);
}
@Bean
public SslBundles sslBundles() {
return new SslBundles() {
@Override
public SslBundle getBundle(String name) throws NoSuchSslBundleException {
return null;
}
@Override
public void addBundleUpdateHandler(String name, java.util.function.Consumer<SslBundle> updateHandler)
throws NoSuchSslBundleException {
}
};
}
}
}