diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java index 1bcee068f..decce7142 100644 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java +++ b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java @@ -28,14 +28,12 @@ import javax.validation.constraints.NotNull; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.producer.ProducerConfig; -import org.springframework.beans.BeansException; import org.springframework.boot.autoconfigure.kafka.KafkaProperties; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.boot.context.properties.DeprecatedConfigurationProperty; import org.springframework.cloud.stream.binder.HeaderMode; import org.springframework.cloud.stream.binder.ProducerProperties; import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties.CompressionType; -import org.springframework.cloud.stream.config.MergableProperties; import org.springframework.expression.Expression; import org.springframework.util.Assert; import org.springframework.util.ObjectUtils; @@ -619,10 +617,6 @@ public class KafkaBinderConfigurationProperties { private final KafkaProducerProperties kafkaProducerProperties = new KafkaProducerProperties(); - public void merge(MergableProperties mergable) { - this.producerProperties.merge(mergable); - } - public Expression getPartitionKeyExpression() { return this.producerProperties.getPartitionKeyExpression(); } @@ -635,10 +629,6 @@ public class KafkaBinderConfigurationProperties { return this.producerProperties.isPartitioned(); } - public void copyProperties(Object source, Object target) throws BeansException { - this.producerProperties.copyProperties(source, target); - } - public Expression getPartitionSelectorExpression() { return this.producerProperties.getPartitionSelectorExpression(); } diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java index e3d5ab739..831c03131 100644 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java +++ b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java @@ -19,7 +19,6 @@ package org.springframework.cloud.stream.binder.kafka.properties; import java.util.HashMap; import java.util.Map; -import org.springframework.cloud.stream.config.MergableProperties; /** * @author Marius Bogoevici @@ -31,7 +30,7 @@ import org.springframework.cloud.stream.config.MergableProperties; * Thanks to Laszlo Szabo for providing the initial patch for generic property support. *

*/ -public class KafkaConsumerProperties implements MergableProperties { +public class KafkaConsumerProperties { public enum StartOffset { earliest(-2L), diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaExtendedBindingProperties.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaExtendedBindingProperties.java index f5f039650..5541f1f48 100644 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaExtendedBindingProperties.java +++ b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaExtendedBindingProperties.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2018 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. @@ -16,76 +16,22 @@ package org.springframework.cloud.stream.binder.kafka.properties; -import java.util.HashMap; -import java.util.Map; - import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.cloud.stream.binder.AbstractExtendedBindingProperties; import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider; -import org.springframework.cloud.stream.binder.ExtendedBindingProperties; /** * @author Marius Bogoevici * @author Gary Russell * @author Soby Chacko + * @author Oleg Zhurakousky */ @ConfigurationProperties("spring.cloud.stream.kafka") public class KafkaExtendedBindingProperties - implements ExtendedBindingProperties { + extends AbstractExtendedBindingProperties { private static final String DEFAULTS_PREFIX = "spring.cloud.stream.kafka.default"; - private Map bindings = new HashMap<>(); - - public Map getBindings() { - return this.bindings; - } - - public void setBindings(Map bindings) { - this.bindings = bindings; - } - - @Override - public synchronized KafkaConsumerProperties getExtendedConsumerProperties(String channelName) { - if (bindings.containsKey(channelName)) { - if (bindings.get(channelName).getConsumer() != null) { - return bindings.get(channelName).getConsumer(); - } - else { - KafkaConsumerProperties properties = new KafkaConsumerProperties(); - this.bindings.get(channelName).setConsumer(properties); - return properties; - } - } - else { - KafkaConsumerProperties properties = new KafkaConsumerProperties(); - KafkaBindingProperties rbp = new KafkaBindingProperties(); - rbp.setConsumer(properties); - bindings.put(channelName, rbp); - return properties; - } - } - - @Override - public synchronized KafkaProducerProperties getExtendedProducerProperties(String channelName) { - if (bindings.containsKey(channelName)) { - if (bindings.get(channelName).getProducer() != null) { - return bindings.get(channelName).getProducer(); - } - else { - KafkaProducerProperties properties = new KafkaProducerProperties(); - this.bindings.get(channelName).setProducer(properties); - return properties; - } - } - else { - KafkaProducerProperties properties = new KafkaProducerProperties(); - KafkaBindingProperties rbp = new KafkaBindingProperties(); - rbp.setProducer(properties); - bindings.put(channelName, rbp); - return properties; - } - } - @Override public String getDefaultsPrefix() { return DEFAULTS_PREFIX; @@ -95,5 +41,4 @@ public class KafkaExtendedBindingProperties public Class getExtendedPropertiesEntryClass() { return KafkaBindingProperties.class; } - } diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaProducerProperties.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaProducerProperties.java index 3874f18dc..0dc4ce1cc 100644 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaProducerProperties.java +++ b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaProducerProperties.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2017 the original author or authors. + * Copyright 2016-2018 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. @@ -21,7 +21,6 @@ import java.util.Map; import javax.validation.constraints.NotNull; -import org.springframework.cloud.stream.config.MergableProperties; import org.springframework.expression.Expression; /** @@ -29,7 +28,7 @@ import org.springframework.expression.Expression; * @author Henryk Konsek * @author Gary Russell */ -public class KafkaProducerProperties implements MergableProperties { +public class KafkaProducerProperties { private int bufferSize = 16384; diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java index 8ecde8f98..e4cc229f7 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java @@ -38,19 +38,14 @@ import org.apache.kafka.streams.state.KeyValueStore; import org.apache.kafka.streams.state.StoreBuilder; import org.apache.kafka.streams.state.Stores; -import org.springframework.beans.BeanUtils; import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanInitializationException; import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.support.BeanDefinitionRegistry; -import org.springframework.boot.context.properties.bind.Bindable; -import org.springframework.boot.context.properties.bind.PropertySourcesPlaceholdersResolver; -import org.springframework.boot.context.properties.source.ConfigurationPropertySources; import org.springframework.cloud.stream.annotation.Input; import org.springframework.cloud.stream.annotation.StreamListener; -import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider; import org.springframework.cloud.stream.binder.ConsumerProperties; import org.springframework.cloud.stream.binder.kafka.streams.annotations.KafkaStreamsStateStore; import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsConsumerProperties; @@ -62,13 +57,11 @@ import org.springframework.cloud.stream.binding.StreamListenerResultAdapter; import org.springframework.cloud.stream.binding.StreamListenerSetupMethodOrchestrator; import org.springframework.cloud.stream.config.BindingProperties; import org.springframework.cloud.stream.config.BindingServiceProperties; -import org.springframework.cloud.stream.config.MergableProperties; import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.core.MethodParameter; import org.springframework.core.annotation.AnnotationUtils; -import org.springframework.integration.support.utils.IntegrationUtils; import org.springframework.kafka.config.KafkaStreamsConfiguration; import org.springframework.kafka.config.StreamsBuilderFactoryBean; import org.springframework.kafka.core.CleanupConfig; @@ -410,9 +403,6 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene Map streamConfigGlobalProperties = applicationContext.getBean("streamConfigGlobalProperties", Map.class); KafkaStreamsConsumerProperties extendedConsumerProperties = kafkaStreamsExtendedBindingProperties.getExtendedConsumerProperties(inboundName); - //Need to apply the default extended properties here as it is not yet done by the BindingService in the binding lifecycle. - handleExtendedDefaultProperties(kafkaStreamsExtendedBindingProperties, - extendedConsumerProperties); String applicationId = extendedConsumerProperties.getApplicationId(); @@ -444,28 +434,6 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene methodStreamsBuilderFactoryBeanMap.put(method, streamsBuilderX); } - // This method is mostly copied from core. We should refactor the original method in core so that it is publicly available - // as a utility method. This is currently hidden as a private method in BindingService. - private void handleExtendedDefaultProperties(KafkaStreamsExtendedBindingProperties kafkaStreamsExtendedBindingProperties, - MergableProperties extendedProperties) { - String defaultsPrefix = kafkaStreamsExtendedBindingProperties.getDefaultsPrefix(); - - if (defaultsPrefix != null) { - Class extendedPropertiesEntryClass = kafkaStreamsExtendedBindingProperties.getExtendedPropertiesEntryClass(); - if (BinderSpecificPropertiesProvider.class.isAssignableFrom(extendedPropertiesEntryClass)) { - org.springframework.boot.context.properties.bind.Binder extendedPropertiesResolverBinder = - new org.springframework.boot.context.properties.bind.Binder(ConfigurationPropertySources.get(applicationContext.getEnvironment()), - new PropertySourcesPlaceholdersResolver(applicationContext.getEnvironment()), - IntegrationUtils.getConversionService(this.applicationContext.getBeanFactory()), null); - BinderSpecificPropertiesProvider defaultProperties = BeanUtils.instantiateClass(extendedPropertiesEntryClass); - extendedPropertiesResolverBinder.bind(defaultsPrefix, Bindable.ofInstance(defaultProperties)); - - Object binderExtendedProperties = defaultProperties.getConsumer(); - ((MergableProperties)binderExtendedProperties).merge(extendedProperties); - } - } - } - @Override public final void setApplicationContext(ApplicationContext applicationContext) throws BeansException { this.applicationContext = (ConfigurableApplicationContext) applicationContext; diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsExtendedBindingProperties.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsExtendedBindingProperties.java index eca6ba627..23b98ccbb 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsExtendedBindingProperties.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsExtendedBindingProperties.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2017-2018 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. @@ -16,52 +16,19 @@ package org.springframework.cloud.stream.binder.kafka.streams.properties; -import java.util.HashMap; -import java.util.Map; - import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.cloud.stream.binder.AbstractExtendedBindingProperties; import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider; -import org.springframework.cloud.stream.binder.ExtendedBindingProperties; - /** * @author Marius Bogoevici + * @author Oleg Zhurakousky */ @ConfigurationProperties("spring.cloud.stream.kafka.streams") public class KafkaStreamsExtendedBindingProperties - implements ExtendedBindingProperties { + extends AbstractExtendedBindingProperties { private static final String DEFAULTS_PREFIX = "spring.cloud.stream.kafka.streams.default"; - private Map bindings = new HashMap<>(); - - public Map getBindings() { - return this.bindings; - } - - public void setBindings(Map bindings) { - this.bindings = bindings; - } - - @Override - public KafkaStreamsConsumerProperties getExtendedConsumerProperties(String binding) { - if (this.bindings.containsKey(binding) && this.bindings.get(binding).getConsumer() != null) { - return this.bindings.get(binding).getConsumer(); - } - else { - return new KafkaStreamsConsumerProperties(); - } - } - - @Override - public KafkaStreamsProducerProperties getExtendedProducerProperties(String binding) { - if (this.bindings.containsKey(binding) && this.bindings.get(binding).getProducer() != null) { - return this.bindings.get(binding).getProducer(); - } - else { - return new KafkaStreamsProducerProperties(); - } - } - @Override public String getDefaultsPrefix() { return DEFAULTS_PREFIX; diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/ExtendedBindingHandlerMappingsProviderConfiguration.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/ExtendedBindingHandlerMappingsProviderConfiguration.java new file mode 100644 index 000000000..0f11bef40 --- /dev/null +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/ExtendedBindingHandlerMappingsProviderConfiguration.java @@ -0,0 +1,47 @@ +/* + * Copyright 2018 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 + * + * http://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.config; + +import java.util.HashMap; +import java.util.Map; + +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; + +/** + * + * @author Oleg Zhurakousky + * + */ + +@Configuration +public class ExtendedBindingHandlerMappingsProviderConfiguration { + + @Bean + public MappingsProvider kafkaExtendedPropertiesDefaultMappingsProvider() { + return () -> { + Map mappings = new HashMap<>(); + mappings.put(ConfigurationPropertyName.of("spring.cloud.stream.kafka.bindings"), + ConfigurationPropertyName.of("spring.cloud.stream.kafka.default")); + mappings.put(ConfigurationPropertyName.of("spring.cloud.stream.kafka.streams"), + ConfigurationPropertyName.of("spring.cloud.stream.kafka.streams.default")); + return mappings; + }; + } +} diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java index d562e0d85..8efa8cc2f 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java @@ -17,11 +17,14 @@ package org.springframework.cloud.stream.binder.kafka.config; import java.io.IOException; +import java.util.Map; import io.micrometer.core.instrument.MeterRegistry; import io.micrometer.core.instrument.binder.MeterBinder; +import org.springframework.beans.factory.SmartInitializingSingleton; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.config.BeanPostProcessor; import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; @@ -29,6 +32,7 @@ import org.springframework.boot.autoconfigure.context.PropertyPlaceholderAutoCon import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration; import org.springframework.boot.autoconfigure.kafka.KafkaProperties; import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.boot.context.properties.source.ConfigurationPropertyName; import org.springframework.cloud.stream.binder.Binder; import org.springframework.cloud.stream.binder.kafka.KafkaBinderMetrics; import org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder; @@ -74,6 +78,7 @@ public class KafkaBinderConfiguration { @Autowired private KafkaProperties kafkaProperties; + @Bean KafkaBinderConfigurationProperties configurationProperties(KafkaProperties kafkaProperties) { return new KafkaBinderConfigurationProperties(kafkaProperties); diff --git a/spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring.factories b/spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring.factories index 956f0a6ee..0af0626c7 100644 --- a/spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring.factories +++ b/spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring.factories @@ -1,2 +1,3 @@ org.springframework.boot.env.EnvironmentPostProcessor=\ org.springframework.cloud.stream.binder.kafka.KafkaBinderEnvironmentPostProcessor +org.springframework.boot.autoconfigure.EnableAutoConfiguration=org.springframework.cloud.stream.binder.kafka.config.ExtendedBindingHandlerMappingsProviderConfiguration