fix scs sources cannot be appended to Kafka binder context (#31715)

.vscode/cspell.json

@@ -630,7 +630,8 @@
       "filename": "sdk/spring/CHANGELOG.md",
       "words": [
         "AADB",
-        "JPMS"
+        "JPMS",
+        "nimbusds"
       ]
     },
     {

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).
 

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<String, BinderProperties> kafkaBinderPropertyMap = new HashMap<>();
                 kafkaBinderPropertyMap.put(KAKFA_BINDER_DEFAULT_NAME, kafkaBinderSourceProperty);
@@ -39,23 +41,39 @@ 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()));
                     }
                 }
             }
         }
         return bean;
     }
 
-    private String buildKafkaBinderSources(BinderProperties binderProperties) {
+    void configureBinderSources(Map<String, Object> 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<String, Object> readSpringMainPropertiesMap(Map<String, Object> map) {
+        if (map.containsKey("spring")) {
+            Map<String, Object> spring = (Map<String, Object>) map.get("spring");
+            if (spring.containsKey("main")) {
+                return (Map<String, Object>) spring.get("main");
+            } else {
+                LinkedHashMap<String, Object> main = new LinkedHashMap<>();
+                spring.put("main", main);
+                return main;
+            }
+        } else {
+            Map<String, Object> main = new LinkedHashMap<>();
+            Map<String, Object> spring = new LinkedHashMap<>();
+            spring.put("main", main);
+            map.put("spring", spring);
+            return main;
+        }
     }
 }

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<String, Map<String, Object>>) 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<String, Map<String, Object>>) 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"));
     }
 
 

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<String, Object> env = new LinkedHashMap<>();
+        Map<String, Object> mainPropertiesMap = buildSpringMainPropertiesMap(env, null, null, null);
+        assertSame(mainPropertiesMap, ((Map<String, Object>) env.get("spring")).get("main"));
+    }
+
+    @Test
+    void testReadSpringMainPropertiesMapWithSpringProp() {
+        Map<String, Object> env = new LinkedHashMap<>();
+        Map<String, Object> mainPropertiesMap = buildSpringMainPropertiesMap(env, "profiles", "active", "dev");
+
+        assertEquals("dev", ((Map<String, Map<String, Object>>) env.get("spring")).get("profiles").get("active"));
+        assertSame(mainPropertiesMap, ((Map<String, Object>) env.get("spring")).get("main"));
+    }
+
+    @Test
+    void testReadSpringMainPropertiesMapWithMainProp() {
+        Map<String, Object> env = new LinkedHashMap<>();
+        Map<String, Object> mainPropertiesMap = buildSpringMainPropertiesMap(env, "main", "banner-mode", "test");
+
+        assertEquals("test", ((Map<String, Map<String, Object>>) env.get("spring")).get("main").get("banner-mode"));
+        assertSame(mainPropertiesMap, ((Map<String, Object>) env.get("spring")).get("main"));
+    }
+
+    @Test
+    void testReadSpringMainPropertiesMapWithSourcesProp() {
+        Map<String, Object> env = new LinkedHashMap<>();
+        Map<String, Object> mainPropertiesMap = buildSpringMainPropertiesMap(env, "main", "sources", "test");
+
+        assertEquals("test", ((Map<String, Map<String, Object>>) env.get("spring")).get("main").get("sources"));
+        assertSame(mainPropertiesMap, ((Map<String, Object>) env.get("spring")).get("main"));
+    }
+
+    @Test
+    void testConfigureBinderSources() {
+        Map<String, Object> env = new LinkedHashMap<>();
+        Map<String, Object> mainPropertiesMap = buildSpringMainPropertiesMap(env, "main", "sources", "test");
+        bpp.configureBinderSources(mainPropertiesMap);
+        assertEquals(AZURE_KAFKA_SPRING_CLOUD_STREAM_CONFIGURATION_CLASS + ",test", ((Map<String, Map<String, Object>>) 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<String, Map<String, Object>>) env.get("spring")).get("main").get("sources"));
+    }
+
+    @Test
+    void testBindKafkaByDefault() {
+        BindingServiceProperties bindingServiceProperties = new BindingServiceProperties();
+        bpp.postProcessBeforeInitialization(bindingServiceProperties, null);
+        Map<String, Object> env = bindingServiceProperties.getBinders().get("kafka")
+                .getEnvironment();
+        assertEquals(AZURE_KAFKA_SPRING_CLOUD_STREAM_CONFIGURATION_CLASS, ((Map<String, Map<String, Object>>) env.get("spring")).get("main").get("sources"));
+
+    }
+
+    @Test
+    void testBindKafkaByName() {
+        BinderProperties binderProperties = new BinderProperties();
+        Map<String, BinderProperties> binders = new HashMap<>();
+        binders.put("kafka", binderProperties);
+        BindingServiceProperties bindingServiceProperties = new BindingServiceProperties();
+        bindingServiceProperties.setBinders(binders);
+
+        bpp.postProcessBeforeInitialization(bindingServiceProperties, null);
+        Map<String, Object> env = bindingServiceProperties.getBinders().get("kafka")
+                .getEnvironment();
+        assertEquals(AZURE_KAFKA_SPRING_CLOUD_STREAM_CONFIGURATION_CLASS, ((Map<String, Map<String, Object>>) env.get("spring")).get("main").get("sources"));
+    }
+
+    @Test
+    void testBindKafkaByType() {
+        BinderProperties binderProperties = new BinderProperties();
+        Map<String, BinderProperties> binders = new HashMap<>();
+        binders.put("test", binderProperties);
+        binderProperties.setType("kafka");
+        BindingServiceProperties bindingServiceProperties = new BindingServiceProperties();
+        bindingServiceProperties.setBinders(binders);
+
+        bpp.postProcessBeforeInitialization(bindingServiceProperties, null);
+        Map<String, Object> env = bindingServiceProperties.getBinders().get("test")
+                .getEnvironment();
+        assertEquals(AZURE_KAFKA_SPRING_CLOUD_STREAM_CONFIGURATION_CLASS, ((Map<String, Map<String, Object>>) env.get("spring")).get("main").get("sources"));
+    }
+
+    private Map<String, Object> buildSpringMainPropertiesMap(Map<String, Object> env, String secondProperty, String thirdProperty, String value) {
+        if (StringUtils.hasText(secondProperty)) {
+            Map<String, Object> second = new LinkedHashMap<>();
+            second.put(thirdProperty, value);
+            Map<String, Object> first = new LinkedHashMap<>();
+            first.put(secondProperty, second);
+            env.put("spring", first);
+        }
+        return bpp.readSpringMainPropertiesMap(env);
+    }
+
+}