GH-75: Don't Ignore User Config Errors

Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-rabbit/issues/75

Instead of setting `ignoreDeclarationExceptions` in the provisioner's `RabbitAdmin`, catch
the exception in the provisioner.
Use a `DeclarationExceptionEvent` to determine that the exception was thrown in some other
admin and treat the condition as fatal.

spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java

@@ -33,6 +33,7 @@ import org.springframework.amqp.core.FanoutExchange;
 import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.core.TopicExchange;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.core.DeclarationExceptionEvent;
 import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
 import org.springframework.beans.factory.support.DefaultListableBeanFactory;
@@ -45,6 +46,7 @@ import org.springframework.cloud.stream.provisioning.ConsumerDestination;
 import org.springframework.cloud.stream.provisioning.ProducerDestination;
 import org.springframework.cloud.stream.provisioning.ProvisioningException;
 import org.springframework.cloud.stream.provisioning.ProvisioningProvider;
+import org.springframework.context.ApplicationListener;
 import org.springframework.context.support.GenericApplicationContext;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
@@ -55,8 +57,9 @@ import org.springframework.util.StringUtils;
  * @author Soby Chacko
  * @author Gary Russell
  */
-public class RabbitExchangeQueueProvisioner implements ProvisioningProvider<ExtendedConsumerProperties<RabbitConsumerProperties>,
-		ExtendedProducerProperties<RabbitProducerProperties>> {
+public class RabbitExchangeQueueProvisioner implements ApplicationListener<DeclarationExceptionEvent>,
+		ProvisioningProvider<ExtendedConsumerProperties<RabbitConsumerProperties>,
+							ExtendedProducerProperties<RabbitProducerProperties>> {
 
 	private static final AnonymousQueue.Base64UrlNamingStrategy ANONYMOUS_GROUP_NAME_GENERATOR
 			= new AnonymousQueue.Base64UrlNamingStrategy("anonymous.");
@@ -71,13 +74,14 @@ public class RabbitExchangeQueueProvisioner implements ProvisioningProvider<Exte
 
 	private final RabbitAdmin rabbitAdmin;
 
+	private boolean notOurAdminException;
+
 	private final GenericApplicationContext autoDeclareContext = new GenericApplicationContext();
 
 	public RabbitExchangeQueueProvisioner(ConnectionFactory connectionFactory) {
 		this.rabbitAdmin = new RabbitAdmin(connectionFactory);
 		this.autoDeclareContext.refresh();
 		this.rabbitAdmin.setApplicationContext(this.autoDeclareContext);
-		this.rabbitAdmin.setIgnoreDeclarationExceptions(true);
 		this.rabbitAdmin.afterPropertiesSet();
 	}
 
@@ -324,6 +328,15 @@ public class RabbitExchangeQueueProvisioner implements ProvisioningProvider<Exte
 				this.logger.debug("Declaration of queue: " + queue.getName() + " deferred - connection not available");
 			}
 		}
+		catch (RuntimeException e) {
+			if (this.notOurAdminException) {
+				this.notOurAdminException = false;
+				throw e;
+			}
+			if (this.logger.isDebugEnabled()) {
+				this.logger.debug("Declaration of queue: " + queue.getName() + " deferred", e);
+			}
+		}
 		addToAutoDeclareContext(beanName, queue);
 	}
 
@@ -419,6 +432,15 @@ public class RabbitExchangeQueueProvisioner implements ProvisioningProvider<Exte
 						"Declaration of exchange: " + exchange.getName() + " deferred - connection not available");
 			}
 		}
+		catch (RuntimeException e) {
+			if (this.notOurAdminException) {
+				this.notOurAdminException = false;
+				throw e;
+			}
+			if (this.logger.isDebugEnabled()) {
+				this.logger.debug("Declaration of exchange: " + exchange.getName() + " deferred", e);
+			}
+		}
 		addToAutoDeclareContext(rootName + ".exchange", exchange);
 	}
 
@@ -440,6 +462,15 @@ public class RabbitExchangeQueueProvisioner implements ProvisioningProvider<Exte
 						"Declaration of binding: " + rootName + ".binding deferred - connection not available");
 			}
 		}
+		catch (RuntimeException e) {
+			if (this.notOurAdminException) {
+				this.notOurAdminException = false;
+				throw e;
+			}
+			if (this.logger.isDebugEnabled()) {
+				this.logger.debug("Declaration of binding: " + rootName + ".binding deferred", e);
+			}
+		}
 		addToAutoDeclareContext(rootName + ".binding", binding);
 	}
 
@@ -462,6 +493,11 @@ public class RabbitExchangeQueueProvisioner implements ProvisioningProvider<Exte
 		}
 	}
 
+	@Override
+	public void onApplicationEvent(DeclarationExceptionEvent event) {
+		this.notOurAdminException = true; // our admin doesn't have an event publisher
+	}
+
 	private static final class RabbitProducerDestination implements ProducerDestination {
 
 		private final Exchange exchange;

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

@@ -32,6 +32,7 @@ import org.junit.Test;
 import org.junit.rules.TestName;
 import org.mockito.ArgumentCaptor;
 
+import org.springframework.amqp.AmqpIOException;
 import org.springframework.amqp.core.AcknowledgeMode;
 import org.springframework.amqp.core.BindingBuilder;
 import org.springframework.amqp.core.DirectExchange;
@@ -48,7 +49,9 @@ import org.springframework.amqp.support.AmqpHeaders;
 import org.springframework.amqp.support.postprocessor.DelegatingDecompressingPostProcessor;
 import org.springframework.amqp.utils.test.TestUtils;
 import org.springframework.beans.DirectFieldAccessor;
+import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
 import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
+import org.springframework.cloud.stream.binder.BinderException;
 import org.springframework.cloud.stream.binder.BinderHeaders;
 import org.springframework.cloud.stream.binder.Binding;
 import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
@@ -64,6 +67,7 @@ import org.springframework.cloud.stream.binder.rabbit.provisioning.RabbitExchang
 import org.springframework.cloud.stream.binder.test.junit.rabbit.RabbitTestSupport;
 import org.springframework.cloud.stream.config.BindingProperties;
 import org.springframework.context.ApplicationContext;
+import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.Lifecycle;
 import org.springframework.expression.spel.standard.SpelExpression;
 import org.springframework.integration.channel.DirectChannel;
@@ -81,6 +85,7 @@ import org.springframework.retry.support.RetryTemplate;
 import com.rabbitmq.http.client.domain.QueueInfo;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.fail;
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
@@ -1095,6 +1100,43 @@ public class RabbitBinderTests extends
 		this.rabbitAvailableRule.getResource().destroy();
 	}
 
+	@Test
+	public void testBadUserDeclarationsFatal() {
+		RabbitTestBinder binder = getBinder();
+		ConfigurableApplicationContext context = TestUtils.getPropertyValue(binder, "binder.applicationContext",
+				ConfigurableApplicationContext.class);
+		ConfigurableListableBeanFactory bf = context.getBeanFactory();
+		bf.registerSingleton("testBadUserDeclarationsFatal", new Queue("testBadUserDeclarationsFatal", false));
+		bf.registerSingleton("binder", binder);
+		RabbitExchangeQueueProvisioner provisioner = TestUtils.getPropertyValue(binder, "binder.provisioningProvider",
+				RabbitExchangeQueueProvisioner.class);
+		bf.initializeBean(provisioner, "provisioner");
+		bf.registerSingleton("provisioner", provisioner);
+		context.addApplicationListener(provisioner);
+		RabbitAdmin admin = new RabbitAdmin(rabbitAvailableRule.getResource());
+		admin.declareQueue(new Queue("testBadUserDeclarationsFatal"));
+		// reset the connection and configure the "user" admin to auto declare queues...
+		rabbitAvailableRule.getResource().resetConnection();
+		bf.initializeBean(admin, "rabbitAdmin");
+		bf.registerSingleton("rabbitAdmin", admin);
+		admin.afterPropertiesSet();
+		// the mis-configured queue should be fatal
+		Binding<?> binding = null;
+		try {
+			binding = binder.bindConsumer("input", "baddecls", new DirectChannel(), createConsumerProperties());
+			fail("Expected exception");
+		}
+		catch (BinderException e) {
+			assertThat(e.getCause()).isInstanceOf(AmqpIOException.class);
+		}
+		finally {
+			admin.deleteQueue("testBadUserDeclarationsFatal");
+			if (binding != null) {
+				binding.unbind();
+			}
+		}
+	}
+
 	private SimpleMessageListenerContainer verifyContainer(Lifecycle endpoint) {
 		SimpleMessageListenerContainer container;
 		RetryTemplate retry;