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 7cf63d6a1..38813528c 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 @@ -208,6 +208,12 @@ 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); + } } } 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 4a36b03be..7e702d93c 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 @@ -1,5 +1,5 @@ /* - * Copyright 2018-2022 the original author or authors. + * Copyright 2018-2024 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. @@ -197,6 +197,30 @@ public class KafkaBinderConfigurationPropertiesTest { 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(); + 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.location")).isEqualTo( + Paths.get(Files.currentFolder().toString(), "target", "testclient.keystore").toString()); + + final Map mergedConsumerConfiguration = kafkaBinderConfigurationProperties.mergedConsumerConfiguration(); + 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.location")).isEqualTo( + Paths.get(Files.currentFolder().toString(), "target", "testclient.keystore").toString()); } private void createContextWithCertFileHandler(HttpServer server, String path) {