mindol1004/spring-cloud-stream-binder-kafka
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

GH-1152: Property binding in Kafka Streams binder

Browse Source 
Add default mappings provider for Kafka Streams (move kafka streams default mapping to new provider)

Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1152
This commit is contained in:
bono007
2021-09-23 08:08:58 -05:00
committed by Soby Chacko
parent a8c948a6b2
commit a4c38b3453
4 changed files with 135 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/ExtendedBindingHandlerMappingsProviderAutoConfiguration.java Normal file
View File

@@ -0,0 +1,51 @@
/*
* Copyright 2018-2021 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.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder.kafka.streams;
import java.util.HashMap;
import java.util.Map;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.context.properties.source.ConfigurationPropertyName;
import org.springframework.cloud.stream.config.BindingHandlerAdvise.MappingsProvider;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* {@link EnableAutoConfiguration Auto-configuration} for extended binding metadata for Kafka Streams.
*
* @author Chris Bono
* @since 3.2
*/
@Configuration(proxyBeanMethods = false)
public class ExtendedBindingHandlerMappingsProviderAutoConfiguration {
@Bean
public MappingsProvider kafkaStreamsExtendedPropertiesDefaultMappingsProvider() {
return () -> {
Map<ConfigurationPropertyName, ConfigurationPropertyName> mappings = new HashMap<>();
mappings.put(
ConfigurationPropertyName.of("spring.cloud.stream.kafka.streams"),
ConfigurationPropertyName.of("spring.cloud.stream.kafka.streams.default"));
mappings.put(
ConfigurationPropertyName.of("spring.cloud.stream.kafka.streams.bindings"),
ConfigurationPropertyName.of("spring.cloud.stream.kafka.streams.default"));
return mappings;
};
}
}

1
spring-cloud-stream-binder-kafka-streams/src/main/resources/META-INF/spring.factories
View File

@@ -1,4 +1,5 @@
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
org.springframework.cloud.stream.binder.kafka.streams.ExtendedBindingHandlerMappingsProviderAutoConfiguration,\
org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsBinderSupportAutoConfiguration,\
org.springframework.cloud.stream.binder.kafka.streams.function.KafkaStreamsFunctionAutoConfiguration,\
org.springframework.cloud.stream.binder.kafka.streams.endpoint.KafkaStreamsTopologyEndpointAutoConfiguration

83
spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/ExtendedBindingHandlerMappingsProviderAutoConfigurationTests.java Normal file
View File

@@ -0,0 +1,83 @@
/*
* Copyright 2019-2021 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.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder.kafka.streams;
import org.junit.jupiter.api.Test;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsExtendedBindingProperties;
import static org.assertj.core.api.Assertions.assertThat;
/**
* Tests for {@link ExtendedBindingHandlerMappingsProviderAutoConfiguration}.
*/
class ExtendedBindingHandlerMappingsProviderAutoConfigurationTests {
private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
.withUserConfiguration(KafkaStreamsTestApp.class)
.withPropertyValues(
"spring.cloud.stream.kafka.streams.default.consumer.application-id: testApp123",
"spring.cloud.stream.kafka.streams.default.consumer.consumed-as: default-consumer",
"spring.cloud.stream.kafka.streams.default.consumer.materialized-as: default-materializer",
"spring.cloud.stream.kafka.streams.default.producer.produced-as: default-producer",
"spring.cloud.stream.kafka.streams.default.producer.key-serde: default-foo");
@Test
void defaultsUsedWhenNoCustomBindingProperties() {
this.contextRunner.run((context) -> {
assertThat(context)
.hasNotFailed()
.hasSingleBean(KafkaStreamsExtendedBindingProperties.class);
KafkaStreamsExtendedBindingProperties extendedBindingProperties = context.getBean(KafkaStreamsExtendedBindingProperties.class);
assertThat(extendedBindingProperties.getExtendedConsumerProperties("process-in-0"))
.hasFieldOrPropertyWithValue("applicationId", "testApp123")
.hasFieldOrPropertyWithValue("consumedAs", "default-consumer")
.hasFieldOrPropertyWithValue("materializedAs", "default-materializer");
assertThat(extendedBindingProperties.getExtendedProducerProperties("process-out-0"))
.hasFieldOrPropertyWithValue("producedAs", "default-producer")
.hasFieldOrPropertyWithValue("keySerde", "default-foo");
});
}
@Test
void defaultsRespectedWhenCustomBindingProperties() {
this.contextRunner
.withPropertyValues(
"spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.consumed-as: custom-consumer",
"spring.cloud.stream.kafka.streams.bindings.process-out-0.producer.produced-as: custom-producer")
.run((context) -> {
assertThat(context)
.hasNotFailed()
.hasSingleBean(KafkaStreamsExtendedBindingProperties.class);
KafkaStreamsExtendedBindingProperties extendedBindingProperties = context.getBean(KafkaStreamsExtendedBindingProperties.class);
assertThat(extendedBindingProperties.getExtendedConsumerProperties("process-in-0"))
.hasFieldOrPropertyWithValue("applicationId", "testApp123")
.hasFieldOrPropertyWithValue("consumedAs", "custom-consumer")
.hasFieldOrPropertyWithValue("materializedAs", "default-materializer");
assertThat(extendedBindingProperties.getExtendedProducerProperties("process-out-0"))
.hasFieldOrPropertyWithValue("producedAs", "custom-producer")
.hasFieldOrPropertyWithValue("keySerde", "default-foo");
});
}
@EnableAutoConfiguration
static class KafkaStreamsTestApp {
}
}