Support extended default properties

Currenlty, we only support default properties for core producer/consumer (spring.cloud.stream.default.producer|consumer...).
These changes add support for configuring default properties for extended producer and consumer properties.

For example, if the binder type is foo, then this allows the applications to configure default properties across
multiple producer or consumer bindings in the form of spring.cloud.stream.foo.default.producer|consumer.property.

The default prefixes for the extended properties are dictated by the respective binder implementations.

Resolves #1360

Manual merging for Map in MergableProperties

Handling values with defaults in extended producer/consumer properties

spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc

@@ -1445,6 +1445,8 @@ The `<channelName>` represents the name of the channel being configured (for exa
 
 To avoid repetition, Spring Cloud Stream supports setting values for all channels, in the format of `spring.cloud.stream.default.<property>=<value>`.
 
+When it comes to avoiding repetitions for extended binding properties, this format should be used - `spring.cloud.stream.<binder-type>.default.<producer|consumer>.<property>=<value>`.
+
 In what follows, we indicate where we have omitted the `spring.cloud.stream.bindings.<channelName>.` prefix and focus just on the property name, with the understanding that the prefix ise included at runtime.
 
 ==== Common Binding Properties

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ExtendedBindingProperties.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.
@@ -22,10 +22,38 @@ package org.springframework.cloud.stream.binder;
  *
  * @author Marius Bogoevici
  * @author Mark Fisher
+ * @author Soby Chacko
  */
 public interface ExtendedBindingProperties<C, P> {
 
 	C getExtendedConsumerProperties(String channelName);
 
 	P getExtendedProducerProperties(String channelName);
+
+	/**
+	 * Extended binding properties can define a default prefix to place all the extended
+	 * common producer and consumer properties. For example, if the binder type is foo
+	 * it is convenient to specify common extended properties for the producer or consumer
+	 * across multiple bindings in the form of `spring.cloud.stream.foo.default.producer.x=y`
+	 * or `spring.cloud.stream.foo.default.consumer.x=y`.
+	 *
+	 * The binding process will use this defaults prefix to resolve any common extended
+	 * producer and consumer properties.
+	 *
+	 * @return default prefix for extended properties
+	 * @since 2.1.0
+	 */
+	String getDefaultsPrefix();
+
+	/**
+	 *
+	 * Extended properties class against which default extended producer and consumer properties
+	 * are resolved. It is expected that this class has two properties - one called producer
+	 * and another called consumer that contains the extended properties for producer and
+	 * consumer respectively.
+	 *
+	 * @return extended properties class that contains extended producer/consumer properties
+	 * @since 2.1.0
+	 */
+	Class<?> getExtendedPropertiesEntryClass();
 }

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java

@@ -16,6 +16,7 @@
 
 package org.springframework.cloud.stream.binding;
 
+import java.lang.reflect.Field;
 import java.sql.Date;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -28,6 +29,9 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
 import org.springframework.beans.BeanUtils;
+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.binder.Binder;
 import org.springframework.cloud.stream.binder.BinderFactory;
 import org.springframework.cloud.stream.binder.Binding;
@@ -39,8 +43,12 @@ import org.springframework.cloud.stream.binder.PollableConsumerBinder;
 import org.springframework.cloud.stream.binder.PollableSource;
 import org.springframework.cloud.stream.binder.ProducerProperties;
 import org.springframework.cloud.stream.config.BindingServiceProperties;
+import org.springframework.cloud.stream.config.MergableProperties;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.integration.support.utils.IntegrationUtils;
 import org.springframework.scheduling.TaskScheduler;
 import org.springframework.util.CollectionUtils;
+import org.springframework.util.ReflectionUtils;
 import org.springframework.util.StringUtils;
 import org.springframework.validation.DataBinder;
 import org.springframework.validation.beanvalidation.CustomValidatorBean;
@@ -72,20 +80,24 @@ public class BindingService {
 
 	private final BinderFactory binderFactory;
 
+	private final ConfigurableApplicationContext applicationContext;
+
 	public BindingService(
 			BindingServiceProperties bindingServiceProperties,
 			BinderFactory binderFactory) {
-		this(bindingServiceProperties, binderFactory, null);
+		this(bindingServiceProperties, binderFactory, null, null);
 	}
 
 	public BindingService(
 			BindingServiceProperties bindingServiceProperties,
-			BinderFactory binderFactory, TaskScheduler taskScheduler) {
+			BinderFactory binderFactory, TaskScheduler taskScheduler,
+			ConfigurableApplicationContext applicationContext) {
 		this.bindingServiceProperties = bindingServiceProperties;
 		this.binderFactory = binderFactory;
 		this.validator = new CustomValidatorBean();
 		this.validator.afterPropertiesSet();
 		this.taskScheduler = taskScheduler;
+		this.applicationContext = applicationContext;
 	}
 
 	@SuppressWarnings({ "unchecked", "rawtypes" })
@@ -101,6 +113,12 @@ public class BindingService {
 			ExtendedConsumerProperties extendedConsumerProperties = new ExtendedConsumerProperties(
 					extension);
 			BeanUtils.copyProperties(consumerProperties, extendedConsumerProperties);
+
+			if (MergableProperties.class.isAssignableFrom(extendedConsumerProperties.getExtension().getClass())) {
+				handleExtendedDefaultProperties((ExtendedPropertiesBinder) binder,
+						(MergableProperties) extendedConsumerProperties.getExtension(), "consumer");
+			}
+
 			consumerProperties = extendedConsumerProperties;
 		}
 
@@ -220,6 +238,11 @@ public class BindingService {
 			ExtendedProducerProperties extendedProducerProperties = new ExtendedProducerProperties<>(
 					extension);
 			BeanUtils.copyProperties(producerProperties, extendedProducerProperties);
+
+			if (MergableProperties.class.isAssignableFrom(extendedProducerProperties.getExtension().getClass())) {
+				handleExtendedDefaultProperties((ExtendedPropertiesBinder) binder,
+						(MergableProperties) extendedProducerProperties.getExtension(), "producer");
+			}
 			producerProperties = extendedProducerProperties;
 		}
 		validate(producerProperties);
@@ -228,6 +251,30 @@ public class BindingService {
 		return binding;
 	}
 
+	private void handleExtendedDefaultProperties(ExtendedPropertiesBinder binder, MergableProperties extendedProperties, String filedName) {
+		String defaultsPrefix = binder.getDefaultsPrefix();
+		Class<?> extendedPropertiesEntryClass = binder.getExtendedPropertiesEntryClass();
+
+		if (defaultsPrefix != null && extendedPropertiesEntryClass != null) {
+
+			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);
+			Object defaultProperties = BeanUtils.instantiateClass(extendedPropertiesEntryClass);
+			extendedPropertiesResolverBinder.bind(defaultsPrefix, Bindable.ofInstance(defaultProperties));
+
+			Field extendedPropertyField = ReflectionUtils.findField(defaultProperties.getClass(), filedName);
+			if (extendedPropertyField != null) {
+				extendedPropertyField.setAccessible(true);
+				Object extendedProducerObject = ReflectionUtils.getField(extendedPropertyField, defaultProperties);
+				if (extendedProducerObject != null) {
+					((MergableProperties)extendedProducerObject).merge(extendedProperties);
+				}
+			}
+		}
+	}
+
 	@SuppressWarnings("rawtypes")
 	public Object getExtendedProducerProperties(Object output, String outputName) {
 		Binder binder =  getBinder(outputName, output.getClass());

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

@@ -47,6 +47,7 @@ import org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPost
 import org.springframework.cloud.stream.function.StreamFunctionProperties;
 import org.springframework.cloud.stream.micrometer.DestinationPublishingMetricsAutoConfiguration;
 import org.springframework.context.ApplicationListener;
+import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.DependsOn;
@@ -163,9 +164,10 @@ public class BindingServiceConfiguration {
 	// already exists).
 	@ConditionalOnMissingBean(search = SearchStrategy.CURRENT)
 	public BindingService bindingService(BindingServiceProperties bindingServiceProperties,
-			BinderFactory binderFactory, TaskScheduler taskScheduler) {
+										BinderFactory binderFactory, TaskScheduler taskScheduler,
+										ConfigurableApplicationContext applicationContext) {
 
-		return new BindingService(bindingServiceProperties, binderFactory, taskScheduler);
+		return new BindingService(bindingServiceProperties, binderFactory, taskScheduler, applicationContext);
 	}
 
 	@Bean

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

@@ -19,6 +19,7 @@ package org.springframework.cloud.stream.config;
 import java.beans.PropertyDescriptor;
 import java.lang.reflect.Method;
 import java.lang.reflect.Modifier;
+import java.util.Map;
 
 import org.springframework.beans.BeanUtils;
 import org.springframework.beans.BeansException;
@@ -26,22 +27,23 @@ 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.CollectionUtils;
 import org.springframework.util.ObjectUtils;
 
 /**
  * NOT INTENDED FOR PUBLIC USE! Was primarily created to address GH-1359.
  *
+ * @author Oleg Zhurakousky
+ * @author Soby Chacko
  * @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:
-	 *
+	 * <p>
 	 * - 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.
@@ -65,16 +67,28 @@ public interface MergableProperties {
 							Object value = readMethod.invoke(this);
 							if (value != null) {
 								if (value instanceof MergableProperties) {
-									((MergableProperties)value).merge((MergableProperties)readMethod.invoke(mergable));
+									((MergableProperties) value).merge((MergableProperties) readMethod.invoke(mergable));
 								}
 								else {
 									Object v = readMethod.invoke(mergable);
-									if (v == null || (ObjectUtils.isArray(v) && ObjectUtils.isEmpty(v))) {
+									if (v == null || (ObjectUtils.isArray(v) && ObjectUtils.isEmpty(v)) ||
+											isEmptyMapAtDestination(v)) {
 										if (!Modifier.isPublic(writeMethod.getDeclaringClass().getModifiers())) {
 											writeMethod.setAccessible(true);
 										}
 										writeMethod.invoke(mergable, value);
 									}
+									else if (isMergableByMap(v)) {
+										handleMapMerging(value, v);
+									}
+									else if (!ObjectUtils.nullSafeEquals(v, value)) {
+										Object obj = BeanUtils.instantiateClass(this.getClass());
+										Object defaultValue = readMethod.invoke(obj);
+										if (ObjectUtils.nullSafeEquals(v, defaultValue)) {
+											writeMethod.invoke(mergable, value);
+										}
+
+									}
 								}
 							}
 						}
@@ -88,6 +102,27 @@ public interface MergableProperties {
 		}
 	}
 
+	default boolean isEmptyMapAtDestination(Object v) {
+		return Map.class.isAssignableFrom(v.getClass()) && CollectionUtils.isEmpty((Map) v);
+	}
+
+	default boolean isMergableByMap(Object v) {
+		return (Map.class.isAssignableFrom(v.getClass()) && !CollectionUtils.isEmpty((Map) v));
+	}
+
+	@SuppressWarnings("unchecked")
+	default void handleMapMerging(Object value, Object v) {
+		if (value instanceof Map) {
+			Map<Object, Object> sourceMap = (Map) value;
+			for (Object key : sourceMap.keySet()) {
+				Map<Object, Object> targetMap = (Map) v;
+				if (!targetMap.containsKey(key)) {
+					targetMap.put(key, sourceMap.get(key));
+				}
+			}
+		}
+	}
+
 	default void copyProperties(Object source, Object target) throws BeansException {
 
 	}

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2017 the original author or authors.
+ * Copyright 2015-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.
@@ -53,7 +53,15 @@ import org.springframework.cloud.stream.config.BindingServiceConfiguration;
 import org.springframework.cloud.stream.config.BindingServiceProperties;
 import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
 import org.springframework.cloud.stream.reflection.GenericsUtils;
+import org.springframework.cloud.stream.utils.FooExtendedConsumerProperties;
+import org.springframework.cloud.stream.utils.FooExtendedProducerProperties;
 import org.springframework.cloud.stream.utils.MockBinderConfiguration;
+import org.springframework.cloud.stream.utils.MockExtendedBinderConfiguration;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.support.GenericApplicationContext;
+import org.springframework.core.env.ConfigurableEnvironment;
+import org.springframework.core.env.MapPropertySource;
+import org.springframework.core.env.StandardEnvironment;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.MessageChannel;
@@ -350,6 +358,74 @@ public class BindingServiceTests {
 		}
 	}
 
+	@Test
+	public void testExtendedDefaultProducerProperties() {
+		BindingServiceProperties serviceProperties = new BindingServiceProperties();
+		Map<String, BindingProperties> bindingProperties = new HashMap<>();
+		BindingProperties props = new BindingProperties();
+		ProducerProperties producerProperties = new ProducerProperties();
+		props.setDestination("dest");
+		props.setProducer(producerProperties);
+		final String outputChannelName = "output";
+		bindingProperties.put(outputChannelName, props);
+		serviceProperties.setBindings(bindingProperties);
+
+		DefaultBinderFactory binderFactory = createMockExtendedBinderFactory();
+
+		ConfigurableApplicationContext applicationContext = new GenericApplicationContext();
+		ConfigurableEnvironment environment = new StandardEnvironment();
+		Map<String, Object> propertiesToAdd = new HashMap<>();
+		propertiesToAdd.put("spring.cloud.stream.foo.default.producer.extendedProperty", "someFancyExtension");
+		environment.getPropertySources().addLast(new MapPropertySource("extPropertiesConfig", propertiesToAdd));
+		applicationContext.setEnvironment(environment);
+
+		BindingService service = new BindingService(serviceProperties, binderFactory, null, applicationContext);
+		MessageChannel outputChannel = new DirectChannel();
+
+		Binder<MessageChannel, ?, ?> binder = binderFactory.getBinder(null, MessageChannel.class);
+		FooExtendedProducerProperties fooExtendedProducerProperties =
+				(FooExtendedProducerProperties)((ExtendedPropertiesBinder)binder).getExtendedProducerProperties("output");
+		assertThat(fooExtendedProducerProperties.getExtendedProperty()).isNull();
+
+		service.bindProducer(outputChannel, outputChannelName);
+
+		assertThat(fooExtendedProducerProperties.getExtendedProperty()).isEqualTo("someFancyExtension");
+	}
+
+	@Test
+	public void testExtendedDefaultConsumerProperties() {
+		BindingServiceProperties serviceProperties = new BindingServiceProperties();
+		Map<String, BindingProperties> bindingProperties = new HashMap<>();
+		BindingProperties props = new BindingProperties();
+		ConsumerProperties consumerProperties = new ConsumerProperties();
+		props.setDestination("dest");
+		props.setConsumer(consumerProperties);
+		final String inputChannelName = "input";
+		bindingProperties.put(inputChannelName, props);
+		serviceProperties.setBindings(bindingProperties);
+
+		DefaultBinderFactory binderFactory = createMockExtendedBinderFactory();
+
+		ConfigurableApplicationContext applicationContext = new GenericApplicationContext();
+		ConfigurableEnvironment environment = new StandardEnvironment();
+		Map<String, Object> propertiesToAdd = new HashMap<>();
+		propertiesToAdd.put("spring.cloud.stream.foo.default.consumer.extendedProperty", "someFancyExtension");
+		environment.getPropertySources().addLast(new MapPropertySource("extPropertiesConfig", propertiesToAdd));
+		applicationContext.setEnvironment(environment);
+
+		BindingService service = new BindingService(serviceProperties, binderFactory, null, applicationContext);
+		MessageChannel inputChannel = new DirectChannel();
+
+		Binder<MessageChannel, ?, ?> binder = binderFactory.getBinder(null, MessageChannel.class);
+		FooExtendedConsumerProperties fooExtendedConsumerProperties =
+				(FooExtendedConsumerProperties)((ExtendedPropertiesBinder)binder).getExtendedConsumerProperties("input");
+		assertThat(fooExtendedConsumerProperties.getExtendedProperty()).isNull();
+
+		service.bindConsumer(inputChannel, inputChannelName);
+
+		assertThat(fooExtendedConsumerProperties.getExtendedProperty()).isEqualTo("someFancyExtension");
+	}
+
 	@Test
 	public void testConsumerPropertiesValidation() {
 		BindingServiceProperties serviceProperties = new BindingServiceProperties();
@@ -457,7 +533,7 @@ public class BindingServiceTests {
 		Binder binder = binderFactory.getBinder("mock", MessageChannel.class);
 		ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
 		scheduler.initialize();
-		BindingService service = new BindingService(properties, binderFactory, scheduler);
+		BindingService service = new BindingService(properties, binderFactory, scheduler, null);
 		MessageChannel inputChannel = new DirectChannel();
 		final Binding<MessageChannel> mockBinding = Mockito.mock(Binding.class);
 		final CountDownLatch fail = new CountDownLatch(2);
@@ -500,7 +576,7 @@ public class BindingServiceTests {
 		Binder binder = binderFactory.getBinder("mock", MessageChannel.class);
 		ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
 		scheduler.initialize();
-		BindingService service = new BindingService(properties, binderFactory, scheduler);
+		BindingService service = new BindingService(properties, binderFactory, scheduler, null);
 		MessageChannel outputChannel = new DirectChannel();
 		final Binding<MessageChannel> mockBinding = Mockito.mock(Binding.class);
 		final CountDownLatch fail = new CountDownLatch(2);
@@ -535,11 +611,24 @@ public class BindingServiceTests {
 				binderTypeRegistry);
 	}
 
+	private DefaultBinderFactory createMockExtendedBinderFactory() {
+		BinderTypeRegistry binderTypeRegistry = createMockExtendedBinderTypeRegistry();
+		return new DefaultBinderFactory(
+				Collections.singletonMap("mock", new BinderConfiguration("mock", new HashMap<>(), true, true)),
+				binderTypeRegistry);
+	}
+
+
 	private DefaultBinderTypeRegistry createMockBinderTypeRegistry() {
 		return new DefaultBinderTypeRegistry(Collections.singletonMap("mock",
 				new BinderType("mock", new Class[] { MockBinderConfiguration.class })));
 	}
 
+	private DefaultBinderTypeRegistry createMockExtendedBinderTypeRegistry() {
+		return new DefaultBinderTypeRegistry(Collections.singletonMap("mock",
+				new BinderType("mock", new Class[] { MockExtendedBinderConfiguration.class })));
+	}
+
 	private BindingServiceProperties createBindingServiceProperties(HashMap<String, String> properties) {
 		BindingServiceProperties bindingServiceProperties = new BindingServiceProperties();
 		org.springframework.boot.context.properties.bind.Binder propertiesBinder = new org.springframework.boot.context.properties.bind.Binder(new MapConfigurationPropertySource(properties));

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/utils/FooBindingProperties.java

@@ -0,0 +1,43 @@
+/*
+ * 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.utils;
+
+/**
+ * @author Soby Chacko
+ */
+public class FooBindingProperties {
+
+	private FooExtendedProducerProperties producer = new FooExtendedProducerProperties();
+
+	private FooExtendedConsumerProperties consumer = new FooExtendedConsumerProperties();
+
+	public FooExtendedProducerProperties getProducer() {
+		return producer;
+	}
+
+	public void setProducer(FooExtendedProducerProperties producer) {
+		this.producer = producer;
+	}
+
+	public FooExtendedConsumerProperties getConsumer() {
+		return consumer;
+	}
+
+	public void setConsumer(FooExtendedConsumerProperties consumer) {
+		this.consumer = consumer;
+	}
+}

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/utils/FooExtendedConsumerProperties.java

@@ -0,0 +1,35 @@
+/*
+ * 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.utils;
+
+import org.springframework.cloud.stream.config.MergableProperties;
+
+/**
+ * @author Soby Chacko
+ */
+public class FooExtendedConsumerProperties implements MergableProperties {
+
+	String extendedProperty;
+
+	public String getExtendedProperty() {
+		return extendedProperty;
+	}
+
+	public void setExtendedProperty(String extendedProperty) {
+		this.extendedProperty = extendedProperty;
+	}
+}

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/utils/FooExtendedProducerProperties.java

@@ -0,0 +1,35 @@
+/*
+ * 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.utils;
+
+import org.springframework.cloud.stream.config.MergableProperties;
+
+/**
+ * @author Soby Chacko
+ */
+public class FooExtendedProducerProperties implements MergableProperties {
+
+	String extendedProperty;
+
+	public String getExtendedProperty() {
+		return extendedProperty;
+	}
+
+	public void setExtendedProperty(String extendedProperty) {
+		this.extendedProperty = extendedProperty;
+	}
+}

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/utils/MockBinderConfiguration.java

@@ -32,4 +32,5 @@ public class MockBinderConfiguration {
 	public Binder<?, ?, ?> binder() {
 		return Mockito.mock(Binder.class, Mockito.withSettings().defaultAnswer(Mockito.RETURNS_MOCKS));
 	}
+
 }

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/utils/MockExtendedBinderConfiguration.java

@@ -0,0 +1,49 @@
+/*
+ * 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.utils;
+
+import org.mockito.Mockito;
+
+import org.springframework.cloud.stream.binder.Binder;
+import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import static org.mockito.Mockito.when;
+
+/**
+ * @author Soby Chacko
+ */
+@Configuration
+public class MockExtendedBinderConfiguration {
+
+	@Bean
+	public Binder<?, ?, ?> extendedPropertiesBinder() {
+		Binder mock = Mockito.mock(Binder.class, Mockito.withSettings().defaultAnswer(Mockito.RETURNS_MOCKS)
+				.extraInterfaces(ExtendedPropertiesBinder.class));
+		when (((ExtendedPropertiesBinder)mock).getExtendedProducerProperties("output"))
+				.thenReturn(new FooExtendedProducerProperties());
+		when (((ExtendedPropertiesBinder)mock).getExtendedConsumerProperties("input"))
+				.thenReturn(new FooExtendedConsumerProperties());
+		when (((ExtendedPropertiesBinder)mock).getDefaultsPrefix())
+				.thenReturn("spring.cloud.stream.foo.default");
+		when (((ExtendedPropertiesBinder)mock).getExtendedPropertiesEntryClass())
+				.thenReturn(FooBindingProperties.class);
+		return mock;
+	}
+
+}