GH-470 Updated extended properties merge logic

Updated extended properties merge logic based on corresponding core updates (see GH-1503)
Resolves #470

polishing

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();
 		}

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.
  * </p>
  */
-public class KafkaConsumerProperties implements MergableProperties {
+public class KafkaConsumerProperties {
 
 	public enum StartOffset {
 		earliest(-2L),

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<KafkaConsumerProperties, KafkaProducerProperties> {
+	extends AbstractExtendedBindingProperties<KafkaConsumerProperties, KafkaProducerProperties, KafkaBindingProperties> {
 
 	private static final String DEFAULTS_PREFIX = "spring.cloud.stream.kafka.default";
 
-	private Map<String, KafkaBindingProperties> bindings = new HashMap<>();
-
-	public Map<String, KafkaBindingProperties> getBindings() {
-		return this.bindings;
-	}
-
-	public void setBindings(Map<String, KafkaBindingProperties> 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<? extends BinderSpecificPropertiesProvider> getExtendedPropertiesEntryClass() {
 		return KafkaBindingProperties.class;
 	}
-
 }

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;
 

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<String, Object> 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<? extends BinderSpecificPropertiesProvider> 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;

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<KafkaStreamsConsumerProperties, KafkaStreamsProducerProperties> {
+		extends AbstractExtendedBindingProperties<KafkaStreamsConsumerProperties, KafkaStreamsProducerProperties, KafkaStreamsBindingProperties> {
 
 	private static final String DEFAULTS_PREFIX = "spring.cloud.stream.kafka.streams.default";
 
-	private Map<String, KafkaStreamsBindingProperties> bindings = new HashMap<>();
-
-	public Map<String, KafkaStreamsBindingProperties> getBindings() {
-		return this.bindings;
-	}
-
-	public void setBindings(Map<String, KafkaStreamsBindingProperties> 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;

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<ConfigurationPropertyName, ConfigurationPropertyName> 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;
+		};
+	}
+}

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

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