From 4f69f05b828b946a418fb37def88abc751c75b19 Mon Sep 17 00:00:00 2001 From: srinivasboini Date: Fri, 9 Feb 2024 05:43:29 +0800 Subject: [PATCH] GH-2882: Schema registry SSL config propagation Fixes: #2882 * Schema registry SSL config propagation to producer and consumer configurations * Verify with tests --- .../KafkaBinderConfigurationProperties.java | 28 +++++---- ...afkaBinderConfigurationPropertiesTest.java | 58 ++++++++++++++----- 2 files changed, 60 insertions(+), 26 deletions(-) diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java index 2dd9d2e1c..7affc351b 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java @@ -62,6 +62,7 @@ import org.springframework.util.StringUtils; * @author Chukwubuikem Ume-Ugwa * @author Nico Heller * @author Norbert Gyurian + * @author Boini Srinivas */ public class KafkaBinderConfigurationProperties { @@ -144,6 +145,12 @@ public class KafkaBinderConfigurationProperties { */ private boolean enableObservation; + + /** + * Schema registry ssl configuration properties. + */ + private final String[] schemaRegistryProperties = new String[]{"schema.registry.url", "schema.registry.ssl.keystore.location", "schema.registry.ssl.keystore.password", "schema.registry.ssl.truststore.location", "schema.registry.ssl.truststore.password", "schema.registry.ssl.key.password"}; + /** * Earlier, @Autowired on this constructor was necessary for all the properties to be discovered * and bound when running as a native application. However, now that Spring Boot fixed the underlying @@ -205,12 +212,6 @@ public class KafkaBinderConfigurationProperties { final String fileSystemLocation = moveCertToFileSystem(storeLocation, this.certificateStoreDirectory); // Overriding the value with absolute filesystem path. this.configuration.put(storeProperty, fileSystemLocation); - // Provide schema-registry properties as producer and consumer properties - // for potential usage by specific serializers/deserializers downstream - if (storeProperty.startsWith("schema.registry")) { - this.producerProperties.put(storeProperty, fileSystemLocation); - this.consumerProperties.put(storeProperty, fileSystemLocation); - } } } @@ -384,11 +385,13 @@ public class KafkaBinderConfigurationProperties { public Map mergedConsumerConfiguration() { Map consumerConfiguration = new HashMap<>(this.kafkaProperties.buildConsumerProperties(null)); // Copy configured binder properties that apply to consumers + // allow schema registry properties to be propagated to consumer configuration for (Map.Entry configurationEntry : this.configuration - .entrySet()) { - if (ConsumerConfig.configNames().contains(configurationEntry.getKey())) { + .entrySet()) { + if (ConsumerConfig.configNames().contains(configurationEntry.getKey()) + || ObjectUtils.containsElement(schemaRegistryProperties, configurationEntry.getKey())) { consumerConfiguration.put(configurationEntry.getKey(), - configurationEntry.getValue()); + configurationEntry.getValue()); } } consumerConfiguration.putAll(this.consumerProperties); @@ -409,10 +412,11 @@ public class KafkaBinderConfigurationProperties { Map producerConfiguration = new HashMap<>(this.kafkaProperties.buildProducerProperties(null)); // Copy configured binder properties that apply to producers for (Map.Entry configurationEntry : this.configuration - .entrySet()) { - if (ProducerConfig.configNames().contains(configurationEntry.getKey())) { + .entrySet()) { + if (ProducerConfig.configNames().contains(configurationEntry.getKey()) + || ObjectUtils.containsElement(schemaRegistryProperties, configurationEntry.getKey())) { producerConfiguration.put(configurationEntry.getKey(), - configurationEntry.getValue()); + configurationEntry.getValue()); } } producerConfiguration.putAll(this.producerProperties); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/test/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationPropertiesTest.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/test/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationPropertiesTest.java index 30d01b0a1..81af28e86 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/test/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationPropertiesTest.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/test/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationPropertiesTest.java @@ -185,8 +185,9 @@ class KafkaBinderConfigurationPropertiesTest { void certificateFilesAreMovedForSchemaRegistryConfiguration() { KafkaProperties kafkaProperties = new KafkaProperties(); KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties = - new KafkaBinderConfigurationProperties(kafkaProperties); + new KafkaBinderConfigurationProperties(kafkaProperties); final Map configuration = kafkaBinderConfigurationProperties.getConfiguration(); + configuration.put("schema.registry.ssl.truststore.location", "classpath:testclient.truststore"); configuration.put("schema.registry.ssl.keystore.location", "classpath:testclient.keystore"); kafkaBinderConfigurationProperties.setCertificateStoreDirectory("target"); @@ -194,20 +195,8 @@ class KafkaBinderConfigurationPropertiesTest { kafkaBinderConfigurationProperties.getKafkaConnectionString(); assertThat(configuration.get("schema.registry.ssl.truststore.location")).isEqualTo( - Paths.get(Files.currentFolder().toString(), "target", "testclient.truststore").toString()); + Paths.get(Files.currentFolder().toString(), "target", "testclient.truststore").toString()); assertThat(configuration.get("schema.registry.ssl.keystore.location")).isEqualTo( - Paths.get(Files.currentFolder().toString(), "target", "testclient.keystore").toString()); - - final Map producerProperties = kafkaBinderConfigurationProperties.getProducerProperties(); - assertThat(producerProperties.get("schema.registry.ssl.truststore.location")).isEqualTo( - Paths.get(Files.currentFolder().toString(), "target", "testclient.truststore").toString()); - assertThat(producerProperties.get("schema.registry.ssl.keystore.location")).isEqualTo( - Paths.get(Files.currentFolder().toString(), "target", "testclient.keystore").toString()); - - final Map consumerProperties = kafkaBinderConfigurationProperties.getConsumerProperties(); - assertThat(consumerProperties.get("schema.registry.ssl.truststore.location")).isEqualTo( - Paths.get(Files.currentFolder().toString(), "target", "testclient.truststore").toString()); - assertThat(consumerProperties.get("schema.registry.ssl.keystore.location")).isEqualTo( Paths.get(Files.currentFolder().toString(), "target", "testclient.keystore").toString()); final Map mergedProducerConfiguration = kafkaBinderConfigurationProperties.mergedProducerConfiguration(); @@ -223,6 +212,47 @@ class KafkaBinderConfigurationPropertiesTest { Paths.get(Files.currentFolder().toString(), "target", "testclient.keystore").toString()); } + @Test + void schemaRegistryPropertiesPropagatedToMergedProducerProperties() { + KafkaProperties kafkaProperties = new KafkaProperties(); + KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties = + new KafkaBinderConfigurationProperties(kafkaProperties); + final Map configuration = kafkaBinderConfigurationProperties.getConfiguration(); + + configuration.put("schema.registry.url", "https://localhost:8081,https://localhost:8082"); + configuration.put("schema.registry.ssl.truststore.location", "classpath:testclient.truststore"); + configuration.put("schema.registry.ssl.keystore.location", "classpath:testclient.keystore"); + configuration.put("schema.registry.ssl.keystore.password", "generated"); + configuration.put("schema.registry.ssl.truststore.password", "generated"); + configuration.put("schema.registry.ssl.key.password", "generated"); + + kafkaBinderConfigurationProperties.setCertificateStoreDirectory("target"); + kafkaBinderConfigurationProperties.getKafkaConnectionString(); + + final Map mergedProducerConfiguration = kafkaBinderConfigurationProperties.mergedProducerConfiguration(); + assertThat(mergedProducerConfiguration.get("schema.registry.url")).isEqualTo("https://localhost:8081,https://localhost:8082"); + assertThat(mergedProducerConfiguration.get("schema.registry.ssl.keystore.location")).isEqualTo( + Paths.get(Files.currentFolder().toString(), "target", "testclient.keystore").toString()); + assertThat(mergedProducerConfiguration.get("schema.registry.ssl.truststore.location")).isEqualTo( + Paths.get(Files.currentFolder().toString(), "target", "testclient.truststore").toString()); + assertThat(mergedProducerConfiguration.get("schema.registry.ssl.keystore.password")).isEqualTo("generated"); + assertThat(mergedProducerConfiguration.get("schema.registry.ssl.truststore.password")).isEqualTo("generated"); + assertThat(mergedProducerConfiguration.get("schema.registry.ssl.key.password")).isEqualTo("generated"); + + + final Map mergedConsumerConfiguration = kafkaBinderConfigurationProperties.mergedConsumerConfiguration(); + assertThat(mergedConsumerConfiguration.get("schema.registry.url")).isEqualTo("https://localhost:8081,https://localhost:8082"); + assertThat(mergedConsumerConfiguration.get("schema.registry.ssl.keystore.location")).isEqualTo( + Paths.get(Files.currentFolder().toString(), "target", "testclient.keystore").toString()); + assertThat(mergedConsumerConfiguration.get("schema.registry.ssl.truststore.location")).isEqualTo( + Paths.get(Files.currentFolder().toString(), "target", "testclient.truststore").toString()); + assertThat(mergedConsumerConfiguration.get("schema.registry.ssl.keystore.password")).isEqualTo("generated"); + assertThat(mergedConsumerConfiguration.get("schema.registry.ssl.truststore.password")).isEqualTo("generated"); + assertThat(mergedConsumerConfiguration.get("schema.registry.ssl.key.password")).isEqualTo("generated"); + + + } + private void createContextWithCertFileHandler(HttpServer server, String path) { server.createContext("/" + path, exchange -> { ClassPathResource ts = new ClassPathResource(path);