Add flag to throw exception on consumer startup failure

The flag can be set in the RMQConnectionFactory. An exception will be
thrown or ignored when a consumer startup fails, depending on the value
of the flag.

This flag allows to maintain backward compatibility, but the behavior in
2.0 will be to throw an exception (see #62).

[#159272755]

References #62

Fixes #53

Author: acogoluegnes

src/main/java/com/rabbitmq/jms/admin/RMQConnectionFactory.java

@@ -119,6 +119,15 @@ public class RMQConnectionFactory implements ConnectionFactory, Referenceable, S
         public void postProcess(com.rabbitmq.client.ConnectionFactory connectionFactory) { }
     };
 
+    /**
+     * Whether an exception should be thrown or not when consumer startup fails.
+     * <p>
+     * Default is false.
+     *
+     * @since 1.10.0
+     */
+    private boolean throwExceptionOnConsumerStartFailure = false;
+
     /** Default not to use ssl */
     private boolean ssl = false;
     private String tlsProtocol;
@@ -249,6 +258,7 @@ protected Connection createConnection(String username, String password, Connecti
             .setRequeueOnMessageListenerException(requeueOnMessageListenerException)
             .setCleanUpServerNamedQueuesForNonDurableTopicsOnSessionClose(this.cleanUpServerNamedQueuesForNonDurableTopicsOnSessionClose)
             .setAmqpPropertiesCustomiser(amqpPropertiesCustomiser)
+            .setThrowExceptionOnConsumerStartFailure(this.throwExceptionOnConsumerStartFailure)
         );
         conn.setTrustedPackages(this.trustedPackages);
         logger.debug("Connection {} created.", conn);
@@ -888,6 +898,18 @@ public void setAmqpConnectionFactoryPostProcessor(AmqpConnectionFactoryPostProce
         this.amqpConnectionFactoryPostProcessor = amqpConnectionFactoryPostProcessor;
     }
 
+    /**
+     * Whether an exception should be thrown or not when consumer startup fails.
+     * <p>
+     * Default is false.
+     *
+     * @param throwExceptionOnConsumerStartFailure
+     * @since 1.10.0
+     */
+    public void setThrowExceptionOnConsumerStartFailure(boolean throwExceptionOnConsumerStartFailure) {
+        this.throwExceptionOnConsumerStartFailure = throwExceptionOnConsumerStartFailure;
+    }
+
     private interface ConnectionCreator {
         com.rabbitmq.client.Connection create(com.rabbitmq.client.ConnectionFactory cf) throws Exception;
     }

src/main/java/com/rabbitmq/jms/client/ConnectionParams.java

@@ -50,16 +50,27 @@ public class ConnectionParams {
      * If set to true, those queues will be deleted when the session is closed.
      * If set to false, queues will be deleted when the owning connection is closed.
      * Default is false.
+     *
      * @since 1.8.0
      */
     private boolean cleanUpServerNamedQueuesForNonDurableTopicsOnSessionClose = false;
 
     /**
      * Callback to customise properties of outbound AMQP messages.
+     *
      * @since 1.9.0
      */
     private AmqpPropertiesCustomiser amqpPropertiesCustomiser;
 
+    /**
+     * Whether an exception should be thrown or not when consumer startup fails.
+     * <p>
+     * Default is false.
+     *
+     * @since 1.10.0
+     */
+    private boolean throwExceptionOnConsumerStartFailure;
+
     public Connection getRabbitConnection() {
         return rabbitConnection;
     }
@@ -140,4 +151,13 @@ public ConnectionParams setAmqpPropertiesCustomiser(AmqpPropertiesCustomiser amq
         this.amqpPropertiesCustomiser = amqpPropertiesCustomiser;
         return this;
     }
+
+    public ConnectionParams setThrowExceptionOnConsumerStartFailure(boolean throwExceptionOnConsumerStartFailure) {
+        this.throwExceptionOnConsumerStartFailure = throwExceptionOnConsumerStartFailure;
+        return this;
+    }
+
+    public boolean willThrowExceptionOnConsumerStartFailure() {
+        return throwExceptionOnConsumerStartFailure;
+    }
 }

src/main/java/com/rabbitmq/jms/client/MessageListenerConsumer.java

@@ -1,4 +1,4 @@
-/* Copyright (c) 2013 Pivotal Software, Inc. All rights reserved. */
+/* Copyright (c) 2013-2018 Pivotal Software, Inc. All rights reserved. */
 package com.rabbitmq.jms.client;
 
 import java.io.IOException;
@@ -40,6 +40,7 @@ class MessageListenerConsumer implements Consumer, Abortable {
     private final long terminationTimeout;
     private volatile boolean rejecting;
     private final boolean requeueOnMessageListenerException;
+    private final boolean throwExceptionOnStartFailure;
 
     /**
      * Constructor
@@ -49,7 +50,7 @@ class MessageListenerConsumer implements Consumer, Abortable {
      * @param terminationTimeout wait time (in nanoseconds) for cancel to take effect
      */
     public MessageListenerConsumer(RMQMessageConsumer messageConsumer, Channel channel, MessageListener messageListener, long terminationTimeout,
-                boolean requeueOnMessageListenerException) {
+                boolean requeueOnMessageListenerException, boolean throwExceptionOnStartFailure) {
         this.messageConsumer = messageConsumer;
         this.channel = channel;
         this.messageListener = messageListener;
@@ -58,6 +59,7 @@ public MessageListenerConsumer(RMQMessageConsumer messageConsumer, Channel chann
         this.completion = new Completion();  // completed when cancelled.
         this.rejecting = this.messageConsumer.getSession().getConnection().isStopped();
         this.requeueOnMessageListenerException = requeueOnMessageListenerException;
+        this.throwExceptionOnStartFailure = throwExceptionOnStartFailure;
     }
 
     private String getConsTag() {
@@ -227,7 +229,9 @@ public void start() throws Exception {
         } catch (Exception e) {
             this.completion.setComplete();  // just in case someone is waiting on it
             logger.error("basicConsume (consumerTag='{}') threw exception", cT, e);
-            throw new RMQJMSException("Error while starting consumer", e);
+            if (throwExceptionOnStartFailure) {
+                throw new RMQJMSException("Error while starting consumer", e);
+            }
         }
     }
 

src/main/java/com/rabbitmq/jms/client/RMQConnection.java

@@ -1,6 +1,32 @@
 /* Copyright (c) 2013-2018 Pivotal Software, Inc. All rights reserved. */
 package com.rabbitmq.jms.client;
 
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.ShutdownListener;
+import com.rabbitmq.client.ShutdownSignalException;
+import com.rabbitmq.jms.util.RMQJMSException;
+import com.rabbitmq.jms.util.WhiteListObjectInputStream;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionConsumer;
+import javax.jms.ConnectionMetaData;
+import javax.jms.Destination;
+import javax.jms.ExceptionListener;
+import javax.jms.IllegalStateException;
+import javax.jms.InvalidClientIDException;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageProducer;
+import javax.jms.Queue;
+import javax.jms.QueueConnection;
+import javax.jms.QueueSession;
+import javax.jms.ServerSessionPool;
+import javax.jms.Session;
+import javax.jms.Topic;
+import javax.jms.TopicConnection;
+import javax.jms.TopicSession;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -10,18 +36,6 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 
-import javax.jms.*;
-import javax.jms.IllegalStateException;
-
-import com.rabbitmq.jms.util.WhiteListObjectInputStream;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.rabbitmq.client.Channel;
-import com.rabbitmq.client.ShutdownListener;
-import com.rabbitmq.client.ShutdownSignalException;
-import com.rabbitmq.jms.util.RMQJMSException;
-
 /**
  * Implementation of the {@link Connection}, {@link QueueConnection} and {@link TopicConnection} interfaces.
  * A {@link RMQConnection} object holds a list of {@link RMQSession} objects as well as the actual
@@ -102,10 +116,20 @@ public class RMQConnection implements Connection, QueueConnection, TopicConnecti
 
     /**
      * Callback to customise properties of outbound AMQP messages.
+     *
      * @since 1.9.0
      */
     private final AmqpPropertiesCustomiser amqpPropertiesCustomiser;
 
+    /**
+     * Whether an exception should be thrown or not when consumer startup fails.
+     * <p>
+     * Default is false.
+     *
+     * @since 1.10.0
+     */
+    private final boolean throwExceptionOnConsumerStartFailure;
+
     /**
      * Classes in these packages can be transferred via ObjectMessage.
      *
@@ -130,6 +154,7 @@ public RMQConnection(ConnectionParams connectionParams) {
         this.requeueOnMessageListenerException = connectionParams.willRequeueOnMessageListenerException();
         this.cleanUpServerNamedQueuesForNonDurableTopicsOnSessionClose = connectionParams.isCleanUpServerNamedQueuesForNonDurableTopicsOnSessionClose();
         this.amqpPropertiesCustomiser = connectionParams.getAmqpPropertiesCustomiser();
+        this.throwExceptionOnConsumerStartFailure = connectionParams.willThrowExceptionOnConsumerStartFailure();
     }
 
     /**
@@ -179,6 +204,7 @@ public Session createSession(boolean transacted, int acknowledgeMode) throws JMS
             .setRequeueOnMessageListenerException(this.requeueOnMessageListenerException)
             .setCleanUpServerNamedQueuesForNonDurableTopics(this.cleanUpServerNamedQueuesForNonDurableTopicsOnSessionClose)
             .setAmqpPropertiesCustomiser(this.amqpPropertiesCustomiser)
+            .setThrowExceptionOnConsumerStartFailure(this.throwExceptionOnConsumerStartFailure)
         );
         session.setTrustedPackages(this.trustedPackages);
         this.sessions.add(session);

src/main/java/com/rabbitmq/jms/client/RMQMessageConsumer.java

@@ -1,4 +1,4 @@
-/* Copyright (c) 2013 Pivotal Software, Inc. All rights reserved. */
+/* Copyright (c) 2013-2018 Pivotal Software, Inc. All rights reserved. */
 package com.rabbitmq.jms.client;
 
 import java.io.IOException;
@@ -84,6 +84,11 @@ public class RMQMessageConsumer implements MessageConsumer, QueueReceiver, Topic
      */
     private final boolean requeueOnMessageListenerException;
 
+    /**
+     * Whether an exception should be thrown or not when consumer startup fails.
+     */
+    private final boolean throwExceptionOnConsumerStartFailure;
+
     /**
      * Creates a RMQMessageConsumer object. Internal constructor used by {@link RMQSession}
      *
@@ -93,8 +98,10 @@ public class RMQMessageConsumer implements MessageConsumer, QueueReceiver, Topic
      *            unique name.
      * @param paused - true if the connection is {@link javax.jms.Connection#stop}ped, false otherwise.
      * @param requeueOnMessageListenerException true to requeue message on RuntimeException in listener, false otherwise
+     * @param throwExceptionOnConsumerStartFailure true to throw an exception if start-up fails, false otherwise
      */
-    RMQMessageConsumer(RMQSession session, RMQDestination destination, String uuidTag, boolean paused, String messageSelector, boolean requeueOnMessageListenerException) {
+    RMQMessageConsumer(RMQSession session, RMQDestination destination, String uuidTag, boolean paused, String messageSelector, boolean requeueOnMessageListenerException,
+            boolean throwExceptionOnConsumerStartFailure) {
         this.session = session;
         this.destination = destination;
         this.uuidTag = uuidTag;
@@ -104,6 +111,7 @@ public class RMQMessageConsumer implements MessageConsumer, QueueReceiver, Topic
             this.receiveManager.openGate();
         this.autoAck = session.isAutoAck();
         this.requeueOnMessageListenerException = requeueOnMessageListenerException;
+        this.throwExceptionOnConsumerStartFailure = throwExceptionOnConsumerStartFailure;
     }
 
     /**
@@ -202,7 +210,8 @@ private void setNewListenerConsumer(MessageListener messageListener) throws Exce
                                           messageListener,
                                           TimeUnit.MILLISECONDS.toNanos(this.session.getConnection()
                                                                                     .getTerminationTimeout()),
-                                          this.requeueOnMessageListenerException);
+                                          this.requeueOnMessageListenerException,
+                                          this.throwExceptionOnConsumerStartFailure);
             if (this.listenerConsumer.compareAndSet(null, mlConsumer)) {
                 this.abortables.add(mlConsumer);
                 if (!this.getSession().getConnection().isStopped()) {

src/main/java/com/rabbitmq/jms/client/RMQSession.java

@@ -92,7 +92,7 @@ public class RMQSession implements Session, QueueSession, TopicSession {
      * {@link javax.jms.MessageListener} or not.
      * Default is false.
      */
-    private boolean requeueOnMessageListenerException = false;
+    private final boolean requeueOnMessageListenerException;
 
     /**
      * Whether using auto-delete for server-named queues for non-durable topics.
@@ -109,6 +109,13 @@ public class RMQSession implements Session, QueueSession, TopicSession {
      */
     private final AmqpPropertiesCustomiser amqpPropertiesCustomiser;
 
+    /**
+     * Whether an exception should be thrown or not when consumer startup fails.
+     *
+     * @since 1.10.0
+     */
+    private final boolean throwExceptionOnConsumerStartFailure;
+
     /** The main RabbitMQ channel we use under the hood */
     private final Channel channel;
     /** Set to true if close() has been called and completed */
@@ -202,6 +209,7 @@ public RMQSession(SessionParams sessionParams) throws JMSException {
         this.requeueOnMessageListenerException = sessionParams.willRequeueOnMessageListenerException();
         this.cleanUpServerNamedQueuesForNonDurableTopics = sessionParams.isCleanUpServerNamedQueuesForNonDurableTopics();
         this.amqpPropertiesCustomiser = sessionParams.getAmqpPropertiesCustomiser();
+        this.throwExceptionOnConsumerStartFailure = sessionParams.willThrowExceptionOnConsumerStartFailure();
 
         if (transacted) {
             this.acknowledgeMode = Session.SESSION_TRANSACTED;
@@ -688,7 +696,8 @@ private RMQMessageConsumer createConsumerInternal(RMQDestination dest, String uu
                 throw new RMQJMSException("RabbitMQ Exception creating Consumer", x);
             }
         }
-        RMQMessageConsumer consumer = new RMQMessageConsumer(this, dest, consumerTag, getConnection().isStopped(), jmsSelector, this.requeueOnMessageListenerException);
+        RMQMessageConsumer consumer = new RMQMessageConsumer(this, dest, consumerTag, getConnection().isStopped(), jmsSelector,
+            this.requeueOnMessageListenerException, this.throwExceptionOnConsumerStartFailure);
         this.consumers.add(consumer);
         return consumer;
     }

src/main/java/com/rabbitmq/jms/client/SessionParams.java

@@ -44,17 +44,29 @@ public class SessionParams {
      * Whether using auto-delete for server-named queues for non-durable topics.
      * If set to true, those queues will be deleted when the session is closed.
      * If set to false, queues will be deleted when the owning connection is closed.
+     * <p>
      * Default is false.
+     *
      * @since 1.8.0
      */
     private boolean cleanUpServerNamedQueuesForNonDurableTopics = false;
 
     /**
      * Callback to customise properties of outbound AMQP messages.
+     *
      * @since 1.9.0
      */
     private AmqpPropertiesCustomiser amqpPropertiesCustomiser;
 
+    /**
+     * Whether an exception should be thrown or not when consumer startup fails.
+     * <p>
+     * Default is false.
+     *
+     * @since 1.10.0
+     */
+    private boolean throwExceptionOnConsumerStartFailure= false;
+
     public RMQConnection getConnection() {
         return connection;
     }
@@ -135,4 +147,13 @@ public SessionParams setAmqpPropertiesCustomiser(AmqpPropertiesCustomiser amqpPr
         this.amqpPropertiesCustomiser = amqpPropertiesCustomiser;
         return this;
     }
+
+    public SessionParams setThrowExceptionOnConsumerStartFailure(boolean throwExceptionOnConsumerStartFailure) {
+        this.throwExceptionOnConsumerStartFailure = throwExceptionOnConsumerStartFailure;
+        return this;
+    }
+
+    public boolean willThrowExceptionOnConsumerStartFailure() {
+        return throwExceptionOnConsumerStartFailure;
+    }
 }