From 3f9162b8b4b39e0a890377daf76fb1dde845f554 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Thu, 1 Feb 2024 13:35:45 -0500 Subject: [PATCH] GH-2882: Schema registry SSL config propagation 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. --- .../KafkaBinderConfigurationProperties.java | 6 +++++ ...afkaBinderConfigurationPropertiesTest.java | 26 ++++++++++++++++++- 2 files changed, 31 insertions(+), 1 deletion(-) 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 76f1dc9ab..2dd9d2e1c 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 @@ -205,6 +205,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 3c72200eb..30d01b0a1 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-2023 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 @@ 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) {