GH-1359 Added support for property precedence

Ensured that default property only takes affect if the actual binding property is not set.
For example if "spring.cloud.stream.bindings.output.producer.partitionCount=4" and "spring.cloud.stream.default.producer.partitionCount=1" are both set
the actual binidng property (i.e., 4)  should take precedence

Resolves #1359

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java

@@ -20,6 +20,8 @@ import javax.validation.constraints.Min;
 
 import com.fasterxml.jackson.annotation.JsonInclude;
 
+import org.springframework.cloud.stream.config.MergableProperties;
+
 /**
  * Common consumer properties.
  *
@@ -30,7 +32,7 @@ import com.fasterxml.jackson.annotation.JsonInclude;
  * @author Oleg Zhurakousky
  */
 @JsonInclude(JsonInclude.Include.NON_DEFAULT)
-public class ConsumerProperties {
+public class ConsumerProperties implements MergableProperties{
 
 	/**
 	 * The concurrency setting of the consumer. Default: 1.

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ProducerProperties.java

@@ -23,6 +23,7 @@ import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.annotation.JsonInclude.Include;
 import com.fasterxml.jackson.databind.annotation.JsonSerialize;
 
+import org.springframework.cloud.stream.config.MergableProperties;
 import org.springframework.expression.Expression;
 
 /**
@@ -34,7 +35,7 @@ import org.springframework.expression.Expression;
  * @author Oleg Zhurakousky
  */
 @JsonInclude(Include.NON_DEFAULT)
-public class ProducerProperties {
+public class ProducerProperties implements MergableProperties {
 
 	@JsonSerialize(using = ExpressionSerializer.class)
 	private Expression partitionKeyExpression;

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java

@@ -37,7 +37,7 @@ import org.springframework.validation.annotation.Validated;
  */
 @JsonInclude(Include.NON_DEFAULT)
 @Validated
-public class BindingProperties {
+public class BindingProperties implements MergableProperties {
 
 	public static final MimeType DEFAULT_CONTENT_TYPE = MimeTypeUtils.APPLICATION_JSON;
 
@@ -153,5 +153,4 @@ public class BindingProperties {
 		sb.deleteCharAt(sb.lastIndexOf(COMMA));
 		return "BindingProperties{" + sb.toString() + "}";
 	}
-
 }

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/EnvironmentEntryInitializingTreeMap.java

@@ -95,9 +95,10 @@ public class EnvironmentEntryInitializingTreeMap<T> extends AbstractMap<String,
 
 	@Override
 	public T put(String key, T value) {
-		// boot 2 call this first
 		Binder binder = new Binder(ConfigurationPropertySources.get(environment),new PropertySourcesPlaceholdersResolver(environment),this.conversionService, null);
-		binder.bind(defaultsPrefix, Bindable.ofInstance(value));
+		T defaultProperties = BeanUtils.instantiateClass(entryClass);
+		binder.bind(defaultsPrefix, Bindable.ofInstance(defaultProperties));
+		((MergableProperties)defaultProperties).merge((MergableProperties) value);
 		return this.delegate.put(key, value);
 	}
 

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/MergableProperties.java

@@ -0,0 +1,91 @@
+/*
+ * 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.config;
+
+import java.beans.PropertyDescriptor;
+import java.lang.reflect.Method;
+import java.lang.reflect.Modifier;
+
+import org.springframework.beans.BeanUtils;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.FatalBeanException;
+import org.springframework.cloud.stream.binder.ConsumerProperties;
+import org.springframework.cloud.stream.binder.ProducerProperties;
+import org.springframework.util.ClassUtils;
+import org.springframework.util.ObjectUtils;
+
+/**
+ * NOT INTENDED FOR PUBLIC USE! Was primarily created to address GH-1359.
+ *
+ * @see BinderProperties
+ * @see ProducerProperties
+ * @see ConsumerProperties
+ *
+ * @author Oleg Zhurakousky
+ */
+public interface MergableProperties {
+
+	/**
+	 * A variation of {@link BeanUtils#copyProperties(Object, Object)} specifically designed to copy properties using the following rule:
+	 *
+	 * - If source property is null then override with the same from mergable.
+	 * - If source property is an array and it is empty then override with same from mergable.
+	 * - If source property is mergable then merge.
+	 */
+	default void merge(MergableProperties mergable) {
+		for (PropertyDescriptor targetPd : BeanUtils.getPropertyDescriptors(mergable.getClass())) {
+			Method writeMethod = targetPd.getWriteMethod();
+			if (writeMethod != null) {
+				PropertyDescriptor sourcePd = BeanUtils.getPropertyDescriptor(this.getClass(), targetPd.getName());
+				if (sourcePd != null) {
+					Method readMethod = sourcePd.getReadMethod();
+					if (readMethod != null &&
+							ClassUtils.isAssignable(writeMethod.getParameterTypes()[0], readMethod.getReturnType())) {
+						try {
+							if (!Modifier.isPublic(readMethod.getDeclaringClass().getModifiers())) {
+								readMethod.setAccessible(true);
+							}
+							Object value = readMethod.invoke(this);
+							if (value != null) {
+								if (value instanceof MergableProperties) {
+									((MergableProperties)value).merge((MergableProperties)readMethod.invoke(mergable));
+								}
+								else {
+									Object v = readMethod.invoke(mergable);
+									if (v == null || (ObjectUtils.isArray(v) && ObjectUtils.isEmpty(v))) {
+										if (!Modifier.isPublic(writeMethod.getDeclaringClass().getModifiers())) {
+											writeMethod.setAccessible(true);
+										}
+										writeMethod.invoke(mergable, value);
+									}
+								}
+							}
+						}
+						catch (Throwable ex) {
+							throw new FatalBeanException(
+									"Could not copy property '" + targetPd.getName() + "' from source to target", ex);
+						}
+					}
+				}
+			}
+		}
+	}
+
+	default void copyProperties(Object source, Object target) throws BeansException {
+
+	}
+}

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingWithGlobalPropertiesTest.java

@@ -42,7 +42,9 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 		"spring.cloud.stream.default.contentType=application/json",
 		"spring.cloud.stream.bindings.output.destination=ticktock",
 		"spring.cloud.stream.default.producer.requiredGroups=someGroup",
-		"spring.cloud.stream.bindings.output.producer.headerMode=none" })
+		"spring.cloud.stream.default.producer.partitionCount=1",
+		"spring.cloud.stream.bindings.output.producer.headerMode=none",
+		"spring.cloud.stream.bindings.output.producer.partitionCount=4"})
 public class SourceBindingWithGlobalPropertiesTest {
 
 	@Autowired
@@ -53,7 +55,8 @@ public class SourceBindingWithGlobalPropertiesTest {
 		BindingProperties bindingProperties = serviceProperties.getBindingProperties(Source.OUTPUT);
 		Assertions.assertThat(bindingProperties.getContentType()).isEqualTo("application/json");
 		Assertions.assertThat(bindingProperties.getDestination()).isEqualTo("ticktock");
-		Assertions.assertThat(bindingProperties.getProducer().getRequiredGroups()).containsExactly("someGroup");
+		Assertions.assertThat(bindingProperties.getProducer().getRequiredGroups()).containsExactly("someGroup"); // default propagates to producer
+		Assertions.assertThat(bindingProperties.getProducer().getPartitionCount()).isEqualTo(4); // validates binding property takes precedence over default
 		Assertions.assertThat(bindingProperties.getProducer().getHeaderMode()).isEqualTo(HeaderMode.none);
 	}