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: https://github.com/spring-cloud/spring-cloud-stream/issues/2882

When providing schema registry SSL truststore and keystore properties
as classpath URL's under Kafka binder configuration, the transformations
of them into corresponding filesystem URL's are not currenlty propagated
into producerProperties and consumerProperties under binder configuration.
This is ncessary for certain serializer/deserializer to work.
See the related GH issue for more context on this.
This commit is contained in:
Soby Chacko
2024-02-01 13:35:45 -05:00
parent 4a991fd994
commit ea6beb2fde
2 changed files with 31 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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);
}
}
}

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

@@ -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<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();
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<String, Object> 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) {