Kafka requires SslBundles for consumer and producer properties

Author: cppwfs

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.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) {

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.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 Object convert(Map<String, Object> source) {
 
 	@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<>());
 	}
 

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.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.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 ListItemWriter<Map<String, Object>> itemWriter() {
 			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 {
+
+				}
+			};
+		}
+
 	}
 
 }

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.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 @@ private void addNameToReaderList(List<Map<String, Object>> itemReaderList, Strin
 			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 {
+
+				}
+			};
+		}
+
 	}
 
 }