From ea6beb2fde92a728848e619646ed766b1a2bedf8 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 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) {