diff --git a/.vscode/cspell.json b/.vscode/cspell.json index 863a4dd252b3..d210d573c97f 100644 --- a/.vscode/cspell.json +++ b/.vscode/cspell.json @@ -630,7 +630,8 @@ "filename": "sdk/spring/CHANGELOG.md", "words": [ "AADB", - "JPMS" + "JPMS", + "nimbusds" ] }, { diff --git a/sdk/spring/CHANGELOG.md b/sdk/spring/CHANGELOG.md index f7b6398396ef..45904b1ca176 100644 --- a/sdk/spring/CHANGELOG.md +++ b/sdk/spring/CHANGELOG.md @@ -10,6 +10,7 @@ Upgrade Spring Boot dependencies version to 3.0.0-M5 and Spring Cloud dependenci - Fix bug: Duplicated "scope" parameter. [#31191](https://github.com/Azure/azure-sdk-for-java/issues/31191). - Fix bug: NimbusJwtDecoder still uses `RestTemplate()` instead `RestTemplateBuilder` [#31233](https://github.com/Azure/azure-sdk-for-java/issues/31233) - Fix bug: Proxy setting not work in Azure AD B2C web application [31593](https://github.com/Azure/azure-sdk-for-java/issues/31593) +- Fix bug: `spring.main.sources` configuration from Spring Cloud Stream Kafka binder cannot take effect. [#31715](https://github.com/Azure/azure-sdk-for-java/pull/31715) #### Breaking Changes - Delete the artifact `spring-cloud-azure-trace-sleuth`. @@ -491,7 +492,7 @@ This section includes changes in the `spring-cloud-azure-autoconfigure` module. * Property name "spring.cloud.azure.active-directory.graph-base-uri" changed to "spring.cloud.azure.active-directory.profile.environment.microsoft-graph-endpoint". * Property name "spring.cloud.azure.active-directory.graph-membership-uri" changed to "spring.cloud.azure.active-directory.profile.environment.microsoft-graph-endpoint" and "spring.cloud.azure.active-directory.user-group.use-transitive-members". - Change AAD B2C configuration properties to use the namespace for credential and environment properties [#25799](https://github.com/Azure/azure-sdk-for-java/pull/25799). -- Change Event Hubs processor configuration properties `spring.cloud.azure.eventhbs.processor.partition-ownership-expiration-interval` to `spring.cloud.azure.eventhbs.processor.load-balancing.partition-ownership-expiration-interval` [#25851](https://github.com/Azure/azure-sdk-for-java/pull/25851). +- Change Event Hubs processor configuration properties `spring.cloud.azure.eventhubs.processor.partition-ownership-expiration-interval` to `spring.cloud.azure.eventhubs.processor.load-balancing.partition-ownership-expiration-interval` [#25851](https://github.com/Azure/azure-sdk-for-java/pull/25851). - Change Event Hubs configuration properties `spring.cloud.azure.eventhubs.fqdn` to `spring.cloud.azure.eventhubs.fully-qualified-namespace` [#25851](https://github.com/Azure/azure-sdk-for-java/pull/25851). - Rename all `*CP` classes to `*ConfigurationProperties` [#26209](https://github.com/Azure/azure-sdk-for-java/pull/26209). diff --git a/sdk/spring/spring-cloud-azure-autoconfigure/src/main/java/com/azure/spring/cloud/autoconfigure/kafka/BindingServicePropertiesBeanPostProcessor.java b/sdk/spring/spring-cloud-azure-autoconfigure/src/main/java/com/azure/spring/cloud/autoconfigure/kafka/BindingServicePropertiesBeanPostProcessor.java index 1862fb6ae2ac..8229b39f5ccd 100644 --- a/sdk/spring/spring-cloud-azure-autoconfigure/src/main/java/com/azure/spring/cloud/autoconfigure/kafka/BindingServicePropertiesBeanPostProcessor.java +++ b/sdk/spring/spring-cloud-azure-autoconfigure/src/main/java/com/azure/spring/cloud/autoconfigure/kafka/BindingServicePropertiesBeanPostProcessor.java @@ -6,8 +6,10 @@ import org.springframework.beans.factory.config.BeanPostProcessor; import org.springframework.cloud.stream.config.BinderProperties; import org.springframework.cloud.stream.config.BindingServiceProperties; +import org.springframework.util.StringUtils; import java.util.HashMap; +import java.util.LinkedHashMap; import java.util.Map; /** @@ -28,7 +30,7 @@ public Object postProcessBeforeInitialization(Object bean, String beanName) thro BindingServiceProperties bindingServiceProperties = (BindingServiceProperties) bean; if (bindingServiceProperties.getBinders().isEmpty()) { BinderProperties kafkaBinderSourceProperty = new BinderProperties(); - configureBinderSources(kafkaBinderSourceProperty, AzureKafkaSpringCloudStreamConfiguration.AZURE_KAFKA_SPRING_CLOUD_STREAM_CONFIGURATION_CLASS); + configureBinderSources(readSpringMainPropertiesMap(kafkaBinderSourceProperty.getEnvironment())); Map kafkaBinderPropertyMap = new HashMap<>(); kafkaBinderPropertyMap.put(KAKFA_BINDER_DEFAULT_NAME, kafkaBinderSourceProperty); @@ -39,7 +41,7 @@ public Object postProcessBeforeInitialization(Object bean, String beanName) thro if (entry.getKey() != null && entry.getValue() != null && (KAKFA_BINDER_TYPE.equalsIgnoreCase(entry.getValue().getType()) || KAKFA_BINDER_DEFAULT_NAME.equalsIgnoreCase(entry.getKey()))) { - configureBinderSources(entry.getValue(), buildKafkaBinderSources(entry.getValue())); + configureBinderSources(readSpringMainPropertiesMap(entry.getValue().getEnvironment())); } } } @@ -47,15 +49,31 @@ public Object postProcessBeforeInitialization(Object bean, String beanName) thro return bean; } - private String buildKafkaBinderSources(BinderProperties binderProperties) { + void configureBinderSources(Map originalSources) { StringBuilder sources = new StringBuilder(AzureKafkaSpringCloudStreamConfiguration.AZURE_KAFKA_SPRING_CLOUD_STREAM_CONFIGURATION_CLASS); - if (binderProperties.getEnvironment().get(SPRING_MAIN_SOURCES_PROPERTY) != null) { - sources.append("," + binderProperties.getEnvironment().get(SPRING_MAIN_SOURCES_PROPERTY)); + if (StringUtils.hasText((String) originalSources.get("sources"))) { + sources.append("," + originalSources.get("sources")); } - return sources.toString(); + originalSources.put("sources", sources.toString()); } - private void configureBinderSources(BinderProperties binderProperties, String sources) { - binderProperties.getEnvironment().put(SPRING_MAIN_SOURCES_PROPERTY, sources); + @SuppressWarnings("unchecked") + Map readSpringMainPropertiesMap(Map map) { + if (map.containsKey("spring")) { + Map spring = (Map) map.get("spring"); + if (spring.containsKey("main")) { + return (Map) spring.get("main"); + } else { + LinkedHashMap main = new LinkedHashMap<>(); + spring.put("main", main); + return main; + } + } else { + Map main = new LinkedHashMap<>(); + Map spring = new LinkedHashMap<>(); + spring.put("main", main); + map.put("spring", spring); + return main; + } } } diff --git a/sdk/spring/spring-cloud-azure-autoconfigure/src/test/java/com/azure/spring/cloud/autoconfigure/kafka/AzureEventHubsKafkaBinderOAuth2AutoConfigurationTest.java b/sdk/spring/spring-cloud-azure-autoconfigure/src/test/java/com/azure/spring/cloud/autoconfigure/kafka/AzureEventHubsKafkaBinderOAuth2AutoConfigurationTest.java index 278df9dfaf34..0cee3fb45b4b 100644 --- a/sdk/spring/spring-cloud-azure-autoconfigure/src/test/java/com/azure/spring/cloud/autoconfigure/kafka/AzureEventHubsKafkaBinderOAuth2AutoConfigurationTest.java +++ b/sdk/spring/spring-cloud-azure-autoconfigure/src/test/java/com/azure/spring/cloud/autoconfigure/kafka/AzureEventHubsKafkaBinderOAuth2AutoConfigurationTest.java @@ -2,13 +2,14 @@ // Licensed under the MIT License. package com.azure.spring.cloud.autoconfigure.kafka; +import java.util.Map; + import org.junit.jupiter.api.Test; import org.springframework.boot.autoconfigure.AutoConfigurations; import org.springframework.boot.test.context.FilteredClassLoader; import org.springframework.boot.test.context.runner.ApplicationContextRunner; import org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfiguration; import org.springframework.cloud.stream.config.BinderFactoryAutoConfiguration; -import org.springframework.cloud.stream.config.BinderProperties; import org.springframework.cloud.stream.config.BindingServiceProperties; import org.springframework.context.support.ConversionServiceFactoryBean; import org.springframework.integration.support.utils.IntegrationUtils; @@ -28,6 +29,7 @@ class AzureEventHubsKafkaBinderOAuth2AutoConfigurationTest { // Required by the init method of BindingServiceProperties .withBean(IntegrationUtils.INTEGRATION_CONVERSION_SERVICE_BEAN_NAME, ConversionServiceFactoryBean.class, ConversionServiceFactoryBean::new); + private final BindingServicePropertiesBeanPostProcessor bpp = new BindingServicePropertiesBeanPostProcessor(); @Test void shouldNotConfigureWithoutKafkaBinderConfigurationClass() { @@ -69,6 +71,39 @@ void shouldConfigureWhenBinderNameSpecified() { assertThat(context).hasSingleBean(BindingServiceProperties.class); testBinderSources(context.getBean(BindingServiceProperties.class), "kafka", AZURE_KAFKA_SPRING_CLOUD_STREAM_CONFIGURATION_CLASS); + assertEquals("value", context.getBean(BindingServiceProperties.class).getBinders().get("kafka").getEnvironment().get("key")); + }); + } + + @Test + @SuppressWarnings("unchecked") + void shouldConfigureWhenOtherSpringEnvironmentSpecified() { + this.contextRunner + .withPropertyValues("spring.cloud.stream.binders.kafka.environment.spring.profiles.active=value") + .run(context -> { + assertThat(context).hasSingleBean(AzureEventHubsKafkaBinderOAuth2AutoConfiguration.class); + assertThat(context).hasSingleBean(BindingServicePropertiesBeanPostProcessor.class); + assertThat(context).hasSingleBean(BindingServiceProperties.class); + + testBinderSources(context.getBean(BindingServiceProperties.class), "kafka", AZURE_KAFKA_SPRING_CLOUD_STREAM_CONFIGURATION_CLASS); + assertEquals("value", ((Map>) context.getBean(BindingServiceProperties.class).getBinders().get("kafka").getEnvironment().get("spring")) + .get("profiles").get("active")); + }); + } + + @Test + @SuppressWarnings("unchecked") + void shouldConfigureWhenOtherSpringMainEnvironmentSpecified() { + this.contextRunner + .withPropertyValues("spring.cloud.stream.binders.kafka.environment.spring.main.banner-mode=console") + .run(context -> { + assertThat(context).hasSingleBean(AzureEventHubsKafkaBinderOAuth2AutoConfiguration.class); + assertThat(context).hasSingleBean(BindingServicePropertiesBeanPostProcessor.class); + assertThat(context).hasSingleBean(BindingServiceProperties.class); + + testBinderSources(context.getBean(BindingServiceProperties.class), "kafka", AZURE_KAFKA_SPRING_CLOUD_STREAM_CONFIGURATION_CLASS); + assertEquals("console", ((Map>) context.getBean(BindingServiceProperties.class).getBinders().get("kafka").getEnvironment().get("spring")) + .get("main").get("banner-mode")); }); } @@ -107,32 +142,21 @@ void shouldConfigureWithMultipleBinders() { @Test void shouldAppendOriginalSources() { - - new ApplicationContextRunner() - .withConfiguration(AutoConfigurations.of(AzureEventHubsKafkaBinderOAuth2AutoConfiguration.class)) - .withBean(IntegrationUtils.INTEGRATION_CONVERSION_SERVICE_BEAN_NAME, ConversionServiceFactoryBean.class, - ConversionServiceFactoryBean::new) - .withBean(BindingServiceProperties.class, () -> { - BindingServiceProperties bindingServiceProperties = new BindingServiceProperties(); - BinderProperties kafkaBinderSourceProperty = new BinderProperties(); - kafkaBinderSourceProperty.getEnvironment().put(SPRING_MAIN_SOURCES_PROPERTY, "test"); - bindingServiceProperties.getBinders().put("kafka", kafkaBinderSourceProperty); - return bindingServiceProperties; - }) + this.contextRunner + .withPropertyValues("spring.cloud.stream.binders.kafka.environment.spring.main.sources=value") .run(context -> { assertThat(context).hasSingleBean(AzureEventHubsKafkaBinderOAuth2AutoConfiguration.class); assertThat(context).hasSingleBean(BindingServicePropertiesBeanPostProcessor.class); assertThat(context).hasSingleBean(BindingServiceProperties.class); - testBinderSources(context.getBean(BindingServiceProperties.class), "kafka", AZURE_KAFKA_SPRING_CLOUD_STREAM_CONFIGURATION_CLASS + ",test"); + testBinderSources(context.getBean(BindingServiceProperties.class), "kafka", AZURE_KAFKA_SPRING_CLOUD_STREAM_CONFIGURATION_CLASS + ",value"); }); } private void testBinderSources(BindingServiceProperties bindingServiceProperties, String binderName, String binderSources) { assertFalse(bindingServiceProperties.getBinders().isEmpty()); assertNotNull(bindingServiceProperties.getBinders().get(binderName)); - assertEquals(binderSources, - bindingServiceProperties.getBinders().get(binderName).getEnvironment().get(SPRING_MAIN_SOURCES_PROPERTY)); + assertEquals(binderSources, bpp.readSpringMainPropertiesMap(bindingServiceProperties.getBinders().get(binderName).getEnvironment()).get("sources")); } diff --git a/sdk/spring/spring-cloud-azure-autoconfigure/src/test/java/com/azure/spring/cloud/autoconfigure/kafka/BindingServicePropertiesBeanPostProcessorTest.java b/sdk/spring/spring-cloud-azure-autoconfigure/src/test/java/com/azure/spring/cloud/autoconfigure/kafka/BindingServicePropertiesBeanPostProcessorTest.java new file mode 100644 index 000000000000..13e2432c584f --- /dev/null +++ b/sdk/spring/spring-cloud-azure-autoconfigure/src/test/java/com/azure/spring/cloud/autoconfigure/kafka/BindingServicePropertiesBeanPostProcessorTest.java @@ -0,0 +1,121 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. +package com.azure.spring.cloud.autoconfigure.kafka; + +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.Map; + +import org.junit.jupiter.api.Test; + +import org.springframework.cloud.stream.config.BinderProperties; +import org.springframework.cloud.stream.config.BindingServiceProperties; +import org.springframework.util.StringUtils; + +import static com.azure.spring.cloud.autoconfigure.kafka.AzureKafkaSpringCloudStreamConfiguration.AZURE_KAFKA_SPRING_CLOUD_STREAM_CONFIGURATION_CLASS; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertSame; + +@SuppressWarnings("unchecked") +class BindingServicePropertiesBeanPostProcessorTest { + + private final BindingServicePropertiesBeanPostProcessor bpp = new BindingServicePropertiesBeanPostProcessor(); + + @Test + void testReadSpringMainPropertiesMapWithoutOriginalValues() { + Map env = new LinkedHashMap<>(); + Map mainPropertiesMap = buildSpringMainPropertiesMap(env, null, null, null); + assertSame(mainPropertiesMap, ((Map) env.get("spring")).get("main")); + } + + @Test + void testReadSpringMainPropertiesMapWithSpringProp() { + Map env = new LinkedHashMap<>(); + Map mainPropertiesMap = buildSpringMainPropertiesMap(env, "profiles", "active", "dev"); + + assertEquals("dev", ((Map>) env.get("spring")).get("profiles").get("active")); + assertSame(mainPropertiesMap, ((Map) env.get("spring")).get("main")); + } + + @Test + void testReadSpringMainPropertiesMapWithMainProp() { + Map env = new LinkedHashMap<>(); + Map mainPropertiesMap = buildSpringMainPropertiesMap(env, "main", "banner-mode", "test"); + + assertEquals("test", ((Map>) env.get("spring")).get("main").get("banner-mode")); + assertSame(mainPropertiesMap, ((Map) env.get("spring")).get("main")); + } + + @Test + void testReadSpringMainPropertiesMapWithSourcesProp() { + Map env = new LinkedHashMap<>(); + Map mainPropertiesMap = buildSpringMainPropertiesMap(env, "main", "sources", "test"); + + assertEquals("test", ((Map>) env.get("spring")).get("main").get("sources")); + assertSame(mainPropertiesMap, ((Map) env.get("spring")).get("main")); + } + + @Test + void testConfigureBinderSources() { + Map env = new LinkedHashMap<>(); + Map mainPropertiesMap = buildSpringMainPropertiesMap(env, "main", "sources", "test"); + bpp.configureBinderSources(mainPropertiesMap); + assertEquals(AZURE_KAFKA_SPRING_CLOUD_STREAM_CONFIGURATION_CLASS + ",test", ((Map>) env.get("spring")).get("main").get("sources")); + + env.clear(); + mainPropertiesMap = buildSpringMainPropertiesMap(env, "main", "profiles", "active"); + bpp.configureBinderSources(mainPropertiesMap); + assertEquals(AZURE_KAFKA_SPRING_CLOUD_STREAM_CONFIGURATION_CLASS, ((Map>) env.get("spring")).get("main").get("sources")); + } + + @Test + void testBindKafkaByDefault() { + BindingServiceProperties bindingServiceProperties = new BindingServiceProperties(); + bpp.postProcessBeforeInitialization(bindingServiceProperties, null); + Map env = bindingServiceProperties.getBinders().get("kafka") + .getEnvironment(); + assertEquals(AZURE_KAFKA_SPRING_CLOUD_STREAM_CONFIGURATION_CLASS, ((Map>) env.get("spring")).get("main").get("sources")); + + } + + @Test + void testBindKafkaByName() { + BinderProperties binderProperties = new BinderProperties(); + Map binders = new HashMap<>(); + binders.put("kafka", binderProperties); + BindingServiceProperties bindingServiceProperties = new BindingServiceProperties(); + bindingServiceProperties.setBinders(binders); + + bpp.postProcessBeforeInitialization(bindingServiceProperties, null); + Map env = bindingServiceProperties.getBinders().get("kafka") + .getEnvironment(); + assertEquals(AZURE_KAFKA_SPRING_CLOUD_STREAM_CONFIGURATION_CLASS, ((Map>) env.get("spring")).get("main").get("sources")); + } + + @Test + void testBindKafkaByType() { + BinderProperties binderProperties = new BinderProperties(); + Map binders = new HashMap<>(); + binders.put("test", binderProperties); + binderProperties.setType("kafka"); + BindingServiceProperties bindingServiceProperties = new BindingServiceProperties(); + bindingServiceProperties.setBinders(binders); + + bpp.postProcessBeforeInitialization(bindingServiceProperties, null); + Map env = bindingServiceProperties.getBinders().get("test") + .getEnvironment(); + assertEquals(AZURE_KAFKA_SPRING_CLOUD_STREAM_CONFIGURATION_CLASS, ((Map>) env.get("spring")).get("main").get("sources")); + } + + private Map buildSpringMainPropertiesMap(Map env, String secondProperty, String thirdProperty, String value) { + if (StringUtils.hasText(secondProperty)) { + Map second = new LinkedHashMap<>(); + second.put(thirdProperty, value); + Map first = new LinkedHashMap<>(); + first.put(secondProperty, second); + env.put("spring", first); + } + return bpp.readSpringMainPropertiesMap(env); + } + +}