diff --git a/spring-cloud-starter-single-step-batch-job/src/main/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemReaderAutoConfiguration.java b/spring-cloud-starter-single-step-batch-job/src/main/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemReaderAutoConfiguration.java index 8812b544..08e63653 100644 --- a/spring-cloud-starter-single-step-batch-job/src/main/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemReaderAutoConfiguration.java +++ b/spring-cloud-starter-single-step-batch-job/src/main/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemReaderAutoConfiguration.java @@ -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> 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) { diff --git a/spring-cloud-starter-single-step-batch-job/src/main/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemWriterAutoConfiguration.java b/spring-cloud-starter-single-step-batch-job/src/main/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemWriterAutoConfiguration.java index a5a12dcf..629ba6c2 100644 --- a/spring-cloud-starter-single-step-batch-job/src/main/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemWriterAutoConfiguration.java +++ b/spring-cloud-starter-single-step-batch-job/src/main/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemWriterAutoConfiguration.java @@ -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> producerFactory() { + ProducerFactory> producerFactory(SslBundles sslBundles) { Map configs = new HashMap<>(); - configs.putAll(this.kafkaProperties.getProducer().buildProperties()); + configs.putAll(this.kafkaProperties.getProducer().buildProperties(sslBundles)); return new DefaultKafkaProducerFactory<>(configs, null, new JsonSerializer<>()); } diff --git a/spring-cloud-starter-single-step-batch-job/src/test/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemReaderAutoConfigurationTests.java b/spring-cloud-starter-single-step-batch-job/src/test/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemReaderAutoConfigurationTests.java index 93688460..82b6a653 100644 --- a/spring-cloud-starter-single-step-batch-job/src/test/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemReaderAutoConfigurationTests.java +++ b/spring-cloud-starter-single-step-batch-job/src/test/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemReaderAutoConfigurationTests.java @@ -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 updateHandler) + throws NoSuchSslBundleException { + + } + }; + } + } } diff --git a/spring-cloud-starter-single-step-batch-job/src/test/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemWriterTests.java b/spring-cloud-starter-single-step-batch-job/src/test/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemWriterTests.java index 65348e8d..2925eab1 100644 --- a/spring-cloud-starter-single-step-batch-job/src/test/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemWriterTests.java +++ b/spring-cloud-starter-single-step-batch-job/src/test/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemWriterTests.java @@ -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 updateHandler) + throws NoSuchSslBundleException { + + } + }; + } + } }