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

GH-2882: Schema registry SSL config propagation

Browse Source 
Fixes: #2882 

* Schema registry SSL config propagation to producer and consumer configurations
* Verify with tests
This commit is contained in:
srinivasboini
2024-02-09 05:43:29 +08:00
committed by GitHub
parent e903c47dac
commit 4f69f05b82
2 changed files with 60 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java
View File

@@ -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<String, Object> mergedConsumerConfiguration() {
Map<String, Object> 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<String, String> 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<String, Object> producerConfiguration = new HashMap<>(this.kafkaProperties.buildProducerProperties(null));
// Copy configured binder properties that apply to producers
for (Map.Entry<String, String> 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);

58
binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/test/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationPropertiesTest.java
View File

@@ -185,8 +185,9 @@ class KafkaBinderConfigurationPropertiesTest {
void certificateFilesAreMovedForSchemaRegistryConfiguration() {
KafkaProperties kafkaProperties = new KafkaProperties();
KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties =
new KafkaBinderConfigurationProperties(kafkaProperties);
new KafkaBinderConfigurationProperties(kafkaProperties);
final Map<String, String> 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<String, String> 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<String, String> 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<String, Object> 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<String, String> 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<String, Object> 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<String, Object> 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);