Ignore unknown binder types if not used

Fix #972

Do not throw an error when binder configurations are
processed, unless a binder configuration that references
an unknown binder type (i.e. not found on the classpath)
is actually used by an application.

This allows supporting scenarios like
http://docs.spring.io/spring-cloud-dataflow/docs/1.2.0.RELEASE/reference/htmlsingle/#dataflow-multiple-brokers
where a set of binder configurations is passed to a number
of applications that might or might not have a specific
binder type on the classpath. If a specific binder configuration
is not used by the application, it can be ignored.

polishing

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

@@ -28,7 +28,7 @@ import java.util.Properties;
  */
 public class BinderConfiguration {
 
-	private final BinderType binderType;
+	private final String binderType;
 
 	private final Properties properties;
 
@@ -44,7 +44,7 @@ public class BinderConfiguration {
 	 * @param defaultCandidate whether the binder should be considered as a candidate when
 	 * determining a default
 	 */
-	public BinderConfiguration(BinderType binderType, Properties properties, boolean inheritEnvironment,
+	public BinderConfiguration(String binderType, Properties properties, boolean inheritEnvironment,
 			boolean defaultCandidate) {
 		this.binderType = binderType;
 		this.properties = properties;
@@ -52,7 +52,7 @@ public class BinderConfiguration {
 		this.defaultCandidate = defaultCandidate;
 	}
 
-	public BinderType getBinderType() {
+	public String getBinderType() {
 		return binderType;
 	}
 

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

@@ -59,8 +59,12 @@ public class DefaultBinderFactory implements BinderFactory, DisposableBean, Appl
 
 	private volatile String defaultBinder;
 
-	public DefaultBinderFactory(Map<String, BinderConfiguration> binderConfigurations) {
+	private final BinderTypeRegistry binderTypeRegistry;
+
+	public DefaultBinderFactory(Map<String, BinderConfiguration> binderConfigurations,
+			BinderTypeRegistry binderTypeRegistry) {
 		this.binderConfigurations = new HashMap<>(binderConfigurations);
+		this.binderTypeRegistry = binderTypeRegistry;
 	}
 
 	@Override
@@ -163,6 +167,8 @@ public class DefaultBinderFactory implements BinderFactory, DisposableBean, Appl
 			if (binderConfiguration == null) {
 				throw new IllegalStateException("Unknown binder configuration: " + configurationName);
 			}
+			BinderType binderType = this.binderTypeRegistry.get(binderConfiguration.getBinderType());
+			Assert.notNull(binderType, "Binder type " + binderConfiguration.getBinderType() + " is not defined");
 			Properties binderProperties = binderConfiguration.getProperties();
 			// Convert all properties to arguments, so that they receive maximum
 			// precedence
@@ -183,7 +189,7 @@ public class DefaultBinderFactory implements BinderFactory, DisposableBean, Appl
 			args.add("--spring.jmx.default-domain=" + defaultDomain + "binder." + configurationName);
 			args.add("--spring.main.applicationContextClass=" + AnnotationConfigApplicationContext.class.getName());
 			List<Class<?>> configurationClasses = new ArrayList<Class<?>>(
-					Arrays.asList(binderConfiguration.getBinderType().getConfigurationClasses()));
+					Arrays.asList(binderType.getConfigurationClasses()));
 			SpringApplicationBuilder springApplicationBuilder = new SpringApplicationBuilder()
 					.sources(configurationClasses.toArray(new Class<?>[] {})).bannerMode(Mode.OFF).web(false);
 			// If the environment is not customized and a main context is available, we

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

@@ -21,7 +21,7 @@ import java.util.HashMap;
 import java.util.Map;
 
 /**
- * Defult implementation of a {@link BinderTypeRegistry}.
+ * Default implementation of a {@link BinderTypeRegistry}.
  *
  * @author Marius Bogoevici
  */

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

@@ -59,17 +59,9 @@ public class BinderFactoryConfiguration {
 	private static final String SELF_CONTAINED_APP_PROPERTY_NAME = SPRING_CLOUD_STREAM_INTERNAL_PREFIX
 			+ ".selfContained";
 
-	private static final String BINDER_CONFIGURATIONS_BEAN_NAME = "spring.cloud.stream.binderConfigruations";
-
 	@Value("${" + SELF_CONTAINED_APP_PROPERTY_NAME + ":}")
 	private String selfContained;
 
-	@Autowired
-	private BinderTypeRegistry binderTypeRegistry;
-
-	@Autowired
-	private BindingServiceProperties bindingServiceProperties;
-
 	@Autowired(required = false)
 	private Collection<DefaultBinderFactory.Listener> binderFactoryListeners;
 
@@ -93,16 +85,19 @@ public class BinderFactoryConfiguration {
 
 	@Bean
 	@ConditionalOnMissingBean(BinderFactory.class)
-	public DefaultBinderFactory binderFactory() {
-		DefaultBinderFactory binderFactory = new DefaultBinderFactory(getBinderConfigurations());
+	public DefaultBinderFactory binderFactory(BinderTypeRegistry binderTypeRegistry,
+			BindingServiceProperties bindingServiceProperties) {
+		DefaultBinderFactory binderFactory = new DefaultBinderFactory(
+				getBinderConfigurations(binderTypeRegistry, bindingServiceProperties), binderTypeRegistry);
 		binderFactory.setDefaultBinder(bindingServiceProperties.getDefaultBinder());
 		binderFactory.setListeners(binderFactoryListeners);
 		return binderFactory;
 	}
 
-	public Map<String, BinderConfiguration> getBinderConfigurations() {
+	private Map<String, BinderConfiguration> getBinderConfigurations(BinderTypeRegistry binderTypeRegistry,
+			BindingServiceProperties bindingServiceProperties) {
 		Map<String, BinderConfiguration> binderConfigurations = new HashMap<>();
-		Map<String, BinderProperties> declaredBinders = this.bindingServiceProperties.getBinders();
+		Map<String, BinderProperties> declaredBinders = bindingServiceProperties.getBinders();
 		boolean defaultCandidatesExist = false;
 		Iterator<Map.Entry<String, BinderProperties>> binderPropertiesIterator = declaredBinders.entrySet().iterator();
 		while (!defaultCandidatesExist && binderPropertiesIterator.hasNext()) {
@@ -111,9 +106,9 @@ public class BinderFactoryConfiguration {
 		List<String> existingBinderConfigurations = new ArrayList<>();
 		for (Map.Entry<String, BinderProperties> binderEntry : declaredBinders.entrySet()) {
 			BinderProperties binderProperties = binderEntry.getValue();
-			if (this.binderTypeRegistry.get(binderEntry.getKey()) != null) {
+			if (binderTypeRegistry.get(binderEntry.getKey()) != null) {
 				binderConfigurations.put(binderEntry.getKey(),
-						new BinderConfiguration(this.binderTypeRegistry.get(binderEntry.getKey()),
+						new BinderConfiguration(binderEntry.getKey(),
 								binderProperties.getEnvironment(), binderProperties.isInheritEnvironment(),
 								binderProperties.isDefaultCandidate()));
 				existingBinderConfigurations.add(binderEntry.getKey());
@@ -121,10 +116,8 @@ public class BinderFactoryConfiguration {
 			else {
 				Assert.hasText(binderProperties.getType(),
 						"No 'type' property present for custom binder " + binderEntry.getKey());
-				BinderType binderType = this.binderTypeRegistry.get(binderProperties.getType());
-				Assert.notNull(binderType, "Binder type " + binderProperties.getType() + " is not defined");
 				binderConfigurations.put(binderEntry.getKey(),
-						new BinderConfiguration(binderType, binderProperties.getEnvironment(),
+						new BinderConfiguration(binderProperties.getType(), binderProperties.getEnvironment(),
 								binderProperties.isInheritEnvironment(), binderProperties.isDefaultCandidate()));
 				existingBinderConfigurations.add(binderEntry.getKey());
 			}
@@ -135,9 +128,9 @@ public class BinderFactoryConfiguration {
 			}
 		}
 		if (!defaultCandidatesExist) {
-			for (Map.Entry<String, BinderType> binderEntry : this.binderTypeRegistry.getAll().entrySet()) {
+			for (Map.Entry<String, BinderType> binderEntry : binderTypeRegistry.getAll().entrySet()) {
 				if (!existingBinderConfigurations.contains(binderEntry.getKey())) {
-					binderConfigurations.put(binderEntry.getKey(), new BinderConfiguration(binderEntry.getValue(),
+					binderConfigurations.put(binderEntry.getKey(), new BinderConfiguration(binderEntry.getKey(),
 							new Properties(), true, true));
 				}
 			}

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

@@ -29,15 +29,20 @@ import org.mockito.Mockito;
 import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
 
+import org.springframework.beans.MutablePropertyValues;
 import org.springframework.beans.factory.NoSuchBeanDefinitionException;
 import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
+import org.springframework.boot.bind.RelaxedDataBinder;
 import org.springframework.cloud.stream.binder.Binder;
 import org.springframework.cloud.stream.binder.BinderConfiguration;
 import org.springframework.cloud.stream.binder.BinderType;
+import org.springframework.cloud.stream.binder.BinderTypeRegistry;
 import org.springframework.cloud.stream.binder.Binding;
 import org.springframework.cloud.stream.binder.ConsumerProperties;
 import org.springframework.cloud.stream.binder.DefaultBinderFactory;
+import org.springframework.cloud.stream.binder.DefaultBinderTypeRegistry;
 import org.springframework.cloud.stream.binder.ProducerProperties;
+import org.springframework.cloud.stream.config.BinderFactoryConfiguration;
 import org.springframework.cloud.stream.config.BindingProperties;
 import org.springframework.cloud.stream.config.BindingServiceProperties;
 import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
@@ -76,9 +81,7 @@ public class BindingServiceTests {
 		final String inputChannelName = "input";
 		bindingProperties.put(inputChannelName, props);
 		properties.setBindings(bindingProperties);
-		DefaultBinderFactory binderFactory = new DefaultBinderFactory(Collections.singletonMap("mock",
-				new BinderConfiguration(new BinderType("mock", new Class[] { MockBinderConfiguration.class }),
-						new Properties(), true, true)));
+		DefaultBinderFactory binderFactory = createMockBinderFactory();
 		Binder binder = binderFactory.getBinder("mock", MessageChannel.class);
 		BindingService service = new BindingService(properties, binderFactory);
 		MessageChannel inputChannel = new DirectChannel();
@@ -98,6 +101,18 @@ public class BindingServiceTests {
 		binderFactory.destroy();
 	}
 
+	private DefaultBinderFactory createMockBinderFactory() {
+		BinderTypeRegistry binderTypeRegistry = createMockBinderTypeRegistry();
+		return new DefaultBinderFactory(
+				Collections.singletonMap("mock", new BinderConfiguration("mock", new Properties(), true, true)),
+				binderTypeRegistry);
+	}
+
+	private DefaultBinderTypeRegistry createMockBinderTypeRegistry() {
+		return new DefaultBinderTypeRegistry(Collections.singletonMap("mock",
+				new BinderType("mock", new Class[] { MockBinderConfiguration.class })));
+	}
+
 	@Test
 	public void testMultipleConsumerBindings() throws Exception {
 		BindingServiceProperties properties = new BindingServiceProperties();
@@ -109,9 +124,7 @@ public class BindingServiceTests {
 
 		properties.setBindings(bindingProperties);
 
-		DefaultBinderFactory binderFactory = new DefaultBinderFactory(Collections.singletonMap("mock",
-				new BinderConfiguration(new BinderType("mock", new Class[] { MockBinderConfiguration.class }),
-						new Properties(), true, true)));
+		DefaultBinderFactory binderFactory = createMockBinderFactory();
 
 		Binder binder = binderFactory.getBinder("mock", MessageChannel.class);
 		BindingService service = new BindingService(properties,
@@ -161,14 +174,7 @@ public class BindingServiceTests {
 		final String inputChannelName = "input";
 		bindingProperties.put(inputChannelName, props);
 		properties.setBindings(bindingProperties);
-		DefaultBinderFactory binderFactory = new DefaultBinderFactory(
-				Collections
-						.singletonMap("mock",
-								new BinderConfiguration(
-										new BinderType("mock",
-												new Class[] {
-														MockBinderConfiguration.class }),
-										new Properties(), true, true)));
+		DefaultBinderFactory binderFactory = createMockBinderFactory();
 		Binder binder = binderFactory.getBinder("mock", MessageChannel.class);
 		BindingService service = new BindingService(properties,
 				binderFactory);
@@ -193,14 +199,7 @@ public class BindingServiceTests {
 	@Test
 	public void checkDynamicBinding() {
 		BindingServiceProperties properties = new BindingServiceProperties();
-		DefaultBinderFactory binderFactory = new DefaultBinderFactory(
-				Collections
-						.singletonMap("mock",
-								new BinderConfiguration(
-										new BinderType("mock",
-												new Class[] {
-														MockBinderConfiguration.class }),
-										new Properties(), true, true)));
+		DefaultBinderFactory binderFactory = createMockBinderFactory();
 		Binder binder = binderFactory.getBinder("mock", MessageChannel.class);
 		@SuppressWarnings("unchecked")
 		Binding<MessageChannel> mockBinding = Mockito.mock(Binding.class);
@@ -210,8 +209,7 @@ public class BindingServiceTests {
 				any(ProducerProperties.class))).thenReturn(mockBinding);
 		BindingService bindingService = new BindingService(properties, binderFactory);
 		SubscribableChannelBindingTargetFactory bindableSubscribableChannelFactory = new SubscribableChannelBindingTargetFactory(
-				new MessageConverterConfigurer(properties,
-						new CompositeMessageConverterFactory()));
+				new MessageConverterConfigurer(properties, new CompositeMessageConverterFactory()));
 		BinderAwareChannelResolver resolver = new BinderAwareChannelResolver(
 				bindingService, bindableSubscribableChannelFactory,
 				new DynamicDestinationsBindable());
@@ -268,9 +266,7 @@ public class BindingServiceTests {
 		final String outputChannelName = "output";
 		bindingProperties.put(outputChannelName, props);
 		serviceProperties.setBindings(bindingProperties);
-		DefaultBinderFactory binderFactory = new DefaultBinderFactory(Collections.singletonMap("mock",
-				new BinderConfiguration(new BinderType("mock", new Class[] { MockBinderConfiguration.class }),
-						new Properties(), true, true)));
+		DefaultBinderFactory binderFactory = createMockBinderFactory();
 		BindingService service = new BindingService(serviceProperties, binderFactory);
 		MessageChannel outputChannel = new DirectChannel();
 		try {
@@ -294,9 +290,7 @@ public class BindingServiceTests {
 		final String inputChannelName = "input";
 		bindingProperties.put(inputChannelName, props);
 		serviceProperties.setBindings(bindingProperties);
-		DefaultBinderFactory binderFactory = new DefaultBinderFactory(Collections.singletonMap("mock",
-				new BinderConfiguration(new BinderType("mock", new Class[] { MockBinderConfiguration.class }),
-						new Properties(), true, true)));
+		DefaultBinderFactory binderFactory = createMockBinderFactory();
 		BindingService service = new BindingService(serviceProperties,
 				binderFactory);
 		MessageChannel inputChannel = new DirectChannel();
@@ -309,6 +303,73 @@ public class BindingServiceTests {
 		}
 	}
 
+	@Test
+	public void testUnknownBinderOnBindingFailure() {
+		HashMap<String, String> properties = new HashMap<>();
+		properties.put("spring.cloud.stream.bindings.input.destination", "fooInput");
+		properties.put("spring.cloud.stream.bindings.input.binder", "mock");
+		properties.put("spring.cloud.stream.bindings.output.destination", "fooOutput");
+		properties.put("spring.cloud.stream.bindings.output.binder", "mockError");
+		BindingServiceProperties bindingServiceProperties = new BindingServiceProperties();
+		RelaxedDataBinder dataBinder = new RelaxedDataBinder(bindingServiceProperties, "spring.cloud.stream");
+		dataBinder.bind(new MutablePropertyValues(properties));
+		BindingService bindingService = new BindingService(bindingServiceProperties,
+				createMockBinderFactory());
+		bindingService.bindConsumer(new DirectChannel(), "input");
+		try {
+			bindingService.bindProducer(new DirectChannel(), "output");
+			fail("Expected 'Unknown binder configuration'");
+		}
+		catch (IllegalStateException e) {
+			assertThat(e).hasMessageContaining("Unknown binder configuration: mockError");
+		}
+	}
+
+	@Test
+	public void testUnrecognizedBinderAllowedIfNotUsed() {
+		HashMap<String, String> properties = new HashMap<>();
+		properties.put("spring.cloud.stream.bindings.input.destination", "fooInput");
+		properties.put("spring.cloud.stream.bindings.output.destination", "fooOutput");
+		properties.put("spring.cloud.stream.defaultBinder", "mock1");
+		properties.put("spring.cloud.stream.binders.mock1.type", "mock");
+		properties.put("spring.cloud.stream.binders.kafka1.type", "kafka");
+		BindingServiceProperties bindingServiceProperties = new BindingServiceProperties();
+		RelaxedDataBinder dataBinder = new RelaxedDataBinder(bindingServiceProperties, "spring.cloud.stream");
+		dataBinder.bind(new MutablePropertyValues(properties));
+		DefaultBinderFactory binderFactory = new BinderFactoryConfiguration()
+				.binderFactory(createMockBinderTypeRegistry(), bindingServiceProperties);
+		BindingService bindingService = new BindingService(bindingServiceProperties,
+				binderFactory);
+		bindingService.bindConsumer(new DirectChannel(), "input");
+		bindingService.bindProducer(new DirectChannel(), "output");
+	}
+
+	@Test
+	public void testUnrecognizedBinderDisallowedIfUsed() {
+		HashMap<String, String> properties = new HashMap<>();
+		properties.put("spring.cloud.stream.bindings.input.destination", "fooInput");
+		properties.put("spring.cloud.stream.bindings.input.binder", "mock1");
+		properties.put("spring.cloud.stream.bindings.output.destination", "fooOutput");
+		properties.put("spring.cloud.stream.bindings.output.type", "kafka1");
+		properties.put("spring.cloud.stream.binders.mock1.type", "mock");
+		properties.put("spring.cloud.stream.binders.kafka1.type", "kafka");
+		BindingServiceProperties bindingServiceProperties = new BindingServiceProperties();
+		RelaxedDataBinder dataBinder = new RelaxedDataBinder(bindingServiceProperties, "spring.cloud.stream");
+		dataBinder.bind(new MutablePropertyValues(properties));
+		DefaultBinderFactory binderFactory = new BinderFactoryConfiguration()
+				.binderFactory(createMockBinderTypeRegistry(), bindingServiceProperties);
+		BindingService bindingService = new BindingService(bindingServiceProperties,
+				binderFactory);
+		bindingService.bindConsumer(new DirectChannel(), "input");
+		try {
+			bindingService.bindProducer(new DirectChannel(), "output");
+			fail("Expected 'Unknown binder configuration'");
+		}
+		catch (IllegalArgumentException e) {
+			assertThat(e).hasMessageContaining("Binder type kafka is not defined");
+		}
+	}
+
 	@Test
 	public void testResolveBindableType() {
 		Class<?> bindableType = GenericsUtils.getParameterType(FooBinder.class, Binder.class, 0);