Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
yiliuTo merged 13 commits into from
Oct 28, 2022
Merged

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
21ce5b3
fix sources cannot be appended
yiliuTo Oct 25, 2022
beca90d
add changelog
yiliuTo Oct 25, 2022
7efce9c
fix checkstyle
yiliuTo Oct 25, 2022
d912d1f
Merge branch 'main' into fix-binder-source-append
yiliuTo Oct 25, 2022
d59fa73
fix changelog error
yiliuTo Oct 25, 2022
02cbda9
modify the way to read/write spring.main.sources
yiliuTo Oct 27, 2022
53a36de
add tests
yiliuTo Oct 27, 2022
9b195fc
fix checkstyle
yiliuTo Oct 27, 2022
d66f7eb
remove duplicated line
yiliuTo Oct 27, 2022
8c2cf93
supress spotbug error
yiliuTo Oct 27, 2022
4d9a065
fix spotbug
yiliuTo Oct 27, 2022
e868918
fix spotbug
yiliuTo Oct 27, 2022
03f8896
Merge branch 'main' into fix-binder-source-append
yiliuTo Oct 28, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion sdk/spring/CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,8 @@ This section includes changes in `spring-cloud-azure-autoconfigure` module.
- Fix bug: RestOperations is not well configured when jwkResolver is null. [#31218](https://github.com/Azure/azure-sdk-for-java/issues/31218).
- 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: 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)
- Fix Bug: NoClassDefFoundError for JSONArray. [31716](https://github.com/Azure/azure-sdk-for-java/issues/31716)

## 4.4.0 (2022-09-26)
Expand Down
34 changes: 26 additions & 8 deletions ...com/azure/spring/cloud/autoconfigure/kafka/BindingServicePropertiesBeanPostProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

/**
Expand All @@ -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);
Expand All @@ -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;
}
}
}
56 changes: 40 additions & 16 deletions ...pring/cloud/autoconfigure/kafka/AzureEventHubsKafkaBinderOAuth2AutoConfigurationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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() {
Expand Down Expand Up @@ -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
Comment thread
saragluna marked this conversation as resolved.
@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"));
});
}

Expand Down Expand Up @@ -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"));
}


Expand Down
121 changes: 121 additions & 0 deletions ...azure/spring/cloud/autoconfigure/kafka/BindingServicePropertiesBeanPostProcessorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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);
}

}