Trying to avoid the max connection warning

[Jack Tauson]

I am trying to figure out a solution for the following error message without increasing the default connection size from 1000 to 2000 or more in the activemq config files.

Recently I came across the following error when around 1000 messages were sent to the broker with a delay of 5 minutes as shown in the code below.

   

WARN  | Could not accept connection  : Exceeded the maximum number of allowed client connections. See the 'maximumConnections' property on the TCP transport configuration URI in the ActiveMQ configuration file (e.g., activemq.xml) | org.apache.activemq.broker.TransportConnector | ActiveMQ Transport Server Thread Handler: tcp://0.0.0.0:61616?maximumConnections=1000&wireFormat.maxFrameSize=104857600

The following is the code which is listening to ActiveMQ continously and as soon as it sees `COMPLETE`, it sends out an email to the user after generating a file.Otherwise, it goes inside `else`
block and sends out the message to the broker again.

Inside the else block, I want to test by closing the connection after I am done sending the message. So I have closed the connection inside a finally block as shown below. Is this a correct way to go about it?

Also, I am opening a new connection everytime I the else block is executed using the following piece of code:

ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
    				connection = connectionFactory.createConnection();
    				connection.start();

How can I use a single connection and avoid using / creating multiple connections multiple times?

 

 @Component
    public class DownloadConsumer {
    	
    	@Autowired
        private JavaMailSender javaMailSender;
    	
    	// one instance, reuse
        private final CloseableHttpClient httpClient = HttpClients.createDefault();
        
    	Connection connection;
    	        
    	
    	// Working Code with JMS 2.0
    	@JmsListener(destination = "MessageProducerJMSV1")
    		public void processBrokerQueues(String message) throws DaoException, JMSException {
    			
    		
    		 try {
    			
    			RequestDao requestDao = (RequestDao) context.getBean("requestDao");
    			
    			String receivedStatus = requestDao.getRequestStatus(message);
    			
    			
    		     
    			//Retrieve Username from the message to include in an email
    			 String[] parts = message.split("#");
    			 String userName = parts[1].trim();
    			 
    			//Retrieve personnelID from the message to include in the webservice calls
    			
    			 String personnelID = parts[3].trim();
    			
    			
    			
    			
    			//Before sending this message, do the check for COMPLETE or ERROR etc
    			if(receivedStatus.equals("COMPLETE")) {
    				
    				
    				
    				String latestUUID = requestDao.getUUID();
    				
    				logger.info("Received UUID in Controller is as follows! ");
    				logger.info(latestUUID);
    				
    				requestDao.sendMessage(message,latestUUID);
    				logger.info("Received status is COMPLETE! ");
    				logger.info("Sending email to the user! ");
    				String emailMessage = "Dear "+userName+",<p>Your files are ready. </p><p> Thanks,<br/> Jack/p>";
    				String recipientEmail = userName+"@organization.com";
    				
    				
    				
    				
    				/*****************************************************\
    				// START: EMAIL Related Code
    				
    				 *******************************************************/
    				
    				MimeMessage msg = javaMailSender.createMimeMessage();
    				 MimeMessageHelper helper = new MimeMessageHelper(msg, true);
    				 helper.setFrom("[e-mail]");
    				 helper.setTo(recipientEmail);
    				 helper.setSubject("Requested Files !");
    				 helper.setText(emailMessage,true);
    				 
    				 javaMailSender.send(msg);
    				 
    				
    			        
    								
    			}
    			else {
    				
    				
    				// Getting JMS connection from the server and starting it
    				ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
    				connection = connectionFactory.createConnection();
    				connection.start();
    				
    				Session session = connection.createSession(false,
    						Session.AUTO_ACKNOWLEDGE);
    				
    				// Destination represents here our queue 'MessageProducerJMSV1' on the  JMS server
    				Destination destination = session.createQueue(subject);
    				
    				
    				MessageProducer producer = session.createProducer(destination);
    				
    				//Sending message to the queue
    				TextMessage toSendMessage = session.createTextMessage(message);
    				
    				long delay = 300 * 1000;
    				long period =300 * 1000;
    				
    				toSendMessage.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, delay);
    				
    				producer.send(toSendMessage);
    				
    				
    				
    				
    			}
    			
    			}
    			catch(Throwable th){
    				th.printStackTrace();	
    				
    			}
    			finally {
    			
    				connection.close();
    				
    			}
    			
    	     }
       // URL of the JMS server. DEFAULT_BROKER_URL will just mean that JMS server is on localhost
    	private static String url = ActiveMQConnection.DEFAULT_BROKER_URL;
    	private static String subject = "MessageProducerJMSV1"; //Queue Name
    	// default broker URL is : tcp://localhost:61616"
    	
    	private static ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("ApplicationContext.xml");
    	private static final Logger logger = LoggerFactory.getLogger(DownloadConsumer.class);
    	
    	
    
    }

[Stephan van Hulst]

You should use a PooledConnectionFactory, and you should close your connections using try-with-resources, not with try-finally. You're also not using Spring's dependency injection mechanism properly.

@Component
public final class DownloadConsumer {

  private static final String SUBJECT = "MessageProducerJMSV1";

  private static final String REQUEST_COMPLETE_STATUS = "COMPLETE";

  private static final Logger LOGGER = LoggerFactory.getLogger(DownloadConsumer.class);

  @Autowired
  private RequestDao requestDao;

  @Autowired
  private JavaMailSender javaMailSender;

  @Autowired
  private ConnectionFactory connectionFactory;

  @JmsListener(destination = SUBJECT)
  public void processBrokerQueues(String message) throws DaoException, JMSException {
    var requestStatus = requestDao.getRequestStatus(message);
    
    switch (requestStatus) {
      case REQUEST_COMPLETE_STATUS:
        sendEmail(message);
        break;

      default:
        enqueue(message);
        break;
    }
  }

  ...

  private void enqueue(String message) throws JMSException {
    try (
      var connection = connectionFactory.createConnection();
    ) {
      connection.start();

      ...
    }
  }
}

Now the trick is to configure Spring that it creates a PooledConnectionFactory for your application once. You should be able to do it like this:

@Configuration
public final class MessageQueueConfiguration {

  @Bean
  public ConnectionFactory createMessageQueueConnectionFactory() {
    return new PooledConnectionFactory(new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_BROKER_URL));
  }
}

[Tim Holloway]

In purely abstract terms, the best way to avoid running out of connections is the make the connection-handling code run as quickly as possible. If your connection requires a process that takes a long time - and sending an email probably will - then instead of doing all the work in your message-handling thread, make the message handler create a request and add it to a queue so that backend processes can deal with the slow parts at leisure.

To scale up from there, you could have multiple message listeners running under a load-balancer and/or multiple backend server machines.

[Jack Tauson]

Stephan van Hulst wrote:You should use a PooledConnectionFactory, and you should close your connections using try-with-resources, not with try-finally. You're also not using Spring's dependency injection mechanism properly.

@Component
public final class DownloadConsumer {

  private static final String SUBJECT = "MessageProducerJMSV1";

  private static final String REQUEST_COMPLETE_STATUS = "COMPLETE";

  private static final Logger LOGGER = LoggerFactory.getLogger(DownloadConsumer.class);

  @Autowired
  private RequestDao requestDao;

  @Autowired
  private JavaMailSender javaMailSender;

  @Autowired
  private ConnectionFactory connectionFactory;

  @JmsListener(destination = SUBJECT)
  public void processBrokerQueues(String message) throws DaoException, JMSException {
    var requestStatus = requestDao.getRequestStatus(message);
    
    switch (requestStatus) {
      case REQUEST_COMPLETE_STATUS:
        sendEmail(message);
        break;

      default:
        enqueue(message);
        break;
    }
  }

  ...

  private void enqueue(String message) throws JMSException {
    try (
      var connection = connectionFactory.createConnection();
    ) {
      connection.start();

      ...
    }
  }
}

Now the trick is to configure Spring that it creates a PooledConnectionFactory for your application once. You should be able to do it like this:

@Configuration
public final class MessageQueueConfiguration {

  @Bean
  public ConnectionFactory createMessageQueueConnectionFactory() {
    return new PooledConnectionFactory(new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_BROKER_URL));
  }
}

Thanks Stephan.

I tested a solution different than yours and didn't use PooledConnectionFactory as shown below and it's working fine:

@Component
public class DownloadConsumer {

   @Autowired
   private JavaMailSender javaMailSender;

   // one instance, reuse
   private final CloseableHttpClient httpClient = HttpClients.createDefault();

   private static Connection connection;

   // URL of the JMS server. DEFAULT_BROKER_URL will just mean that JMS server is on localhost
   private static String url = ActiveMQConnection.DEFAULT_BROKER_URL;
   private static String subject = "MessageProducerJMSV1"; //Queue Name
   // default broker URL is : tcp://localhost:61616"

   private static ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("ApplicationContext.xml");
   private static final Logger logger = LoggerFactory.getLogger(DownloadConsumer.class);

   // Working Code with JMS 2.0
   @JmsListener(destination = "MessageProducerJMSV1")
   public void processBrokerQueues(String message) throws DaoException, JMSException {

      try {

         RequestDao requestDao = (RequestDao) context.getBean("requestDao");

         String receivedStatus = requestDao.getRequestStatus(message);

         //Retrieve Username from the message to include in an email
         String[] parts = message.split("#");
         String userName = parts[1].trim();

         //Retrieve personnelID from the message to include in the webservice calls

         String personnelID = parts[3].trim();

         //Before sending this message, do the check for COMPLETE or ERROR etc
         if (receivedStatus.equals("COMPLETE")) {

            String latestUUID = requestDao.getUUID();

            logger.info("Received UUID in Controller is as follows! ");
            logger.info(latestUUID);

            requestDao.sendMessage(message, latestUUID);
            logger.info("Received status is COMPLETE! ");
            logger.info("Sending email to the user! ");
            String emailMessage = "Dear " + userName + ",<p>Your files are ready. </p><p> Thanks,<br/> Jack/p>";
            String recipientEmail = userName + "@organization.com";

            /*****************************************************\
             // START: EMAIL Related Code
             *******************************************************/

            MimeMessage msg = javaMailSender.createMimeMessage();
            MimeMessageHelper helper = new MimeMessageHelper(msg, true);
            helper.setFrom("[e-mail]");
            helper.setTo(recipientEmail);
            helper.setSubject("Requested Files !");
            helper.setText(emailMessage, true);

            javaMailSender.send(msg);

         } else {
            try {
               if (connection == null) {
                  createConnection();
               }
               Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

               // Destination represents here our queue 'MessageProducerJMSV1' on the  JMS server
               Destination destination = session.createQueue(subject);

               MessageProducer producer = session.createProducer(destination);

               //Sending message to the queue
               TextMessage toSendMessage = session.createTextMessage(message);

               long delay = 300 * 1000;
               long period = 300 * 1000;

               toSendMessage.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, delay);

               producer.send(toSendMessage);
            } catch (Throwable th) {
               th.printStackTrace();
               
               // if there are any problems close the connection and it will be re-created next time
               if (connection != null) {
                  connection.close();
                  connection = null;
               }
            }
         }
      } catch (Throwable th) {
         th.printStackTrace();
      }
   }
   
   private void createConnection() {
      // Getting JMS connection from the server
      ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
      connection = connectionFactory.createConnection();
   }
}

So basically, the connection is cached  (e.g. in a static variable) and re-used as shown above. It didn't break through until now.

You'll note that in this code there is no finally block to close the connection. That's intentional because the whole point of this code is keep the connection open so that it's not opening & closing the connection to send a single message. The connection is being re-used between invocations. The only time the connection is closed is when a Throwable is caught.

Do you think I would still need to add a finally block, test for open connection and if it's open then close it? I wonder since finally block is always executed, it's going to close the connection after exiting the else block, right?

[Tim Holloway]

As a general rule, a Factory is a service class that's designed to dispense resources efficiently. If the resources are poolable, then often the factory will maintain its own internal pool. For example the DataSource factory that's built into every J(2)EE-compliant web application server.

Also as a general rule, holding resources for indefinite periods of time is stealing reasources from other users. And I won't even begin to consider what could happen if you have a single instance of something and multiple thread users without any synchronization.

Finally, while there are times where storing things in static variables and/or using static methods is essential, more often than not it's going to be trouble sooner or later. I speak from repeated experience, alas. Too often repeated.

Which brings me to my final observation. Just because something "works" isn't good enough. You should be able to confirm with reasonably certainty that it will always work, and not just when you're there watching it. Murphy's Law says that when it doesn't work will be at the worst possible time. If you are wise/experienced/covered with burn scars, you will not only proof your solution, you'll add logging and other de- and anti-bugging code to the solution so that when it does go wrong, you'll know why, what "impossible" data values were involved and how to most expeditiously recover, diagnore and repair.

[Jack Tauson]

Tim Holloway wrote:As a general rule, a Factory is a service class that's designed to dispense resources efficiently. If the resources are poolable, then often the factory will maintain its own internal pool. For example the DataSource factory that's built into every J(2)EE-compliant web application server.

Thanks for the explanation. I will try to test the pooled connection factory approach that Stephan has mentioned in his code above.

And I won't even begin to consider what could happen if you have a single instance of something and multiple thread users without any synchronization.

Do you mean when multiple users are accessing the application at the same time?

[Dave Tolls]

Doesn't have to be multiple users, as in people, just multiple threads.

If more than one thread at a time accesses this code then something will eventually break.
All it would take is for one thread to close the connection before another thread tries to use either it or the Session value from lines 70-78.

[Tim Holloway]

Dave Tolls wrote:Doesn't have to be multiple users, as in people, just multiple threads.

If more than one thread at a time accesses this code then something will eventually break.
All it would take is for one thread to close the connection before another thread tries to use either it or the Session value from lines 70-78.

Or for two threads to try to use the connection at once and thus corrupt the connection's internal state.

[Dave Tolls]

Indeed.
And that can be quite hard to spot.

[Jack Tauson]

Tim Holloway wrote:

Dave Tolls wrote:Doesn't have to be multiple users, as in people, just multiple threads.

If more than one thread at a time accesses this code then something will eventually break.
All it would take is for one thread to close the connection before another thread tries to use either it or the Session value from lines 70-78.

Or for two threads to try to use the connection at once and thus corrupt the connection's internal state.

I see. How about using addressing multiple people accessing the application at the same time by adding synchronized block as shown below:

@Component
public class DownloadConsumer {

   @Autowired
   private JavaMailSender javaMailSender;

   // one instance, reuse
   private final CloseableHttpClient httpClient = HttpClients.createDefault();

   private static Connection connection;

   private static Object connectionLock = new Object();

   // URL of the JMS server. DEFAULT_BROKER_URL will just mean that JMS server is on localhost
   private static String url = ActiveMQConnection.DEFAULT_BROKER_URL;
   private static String subject = "MessageProducerJMSV1"; //Queue Name
   // default broker URL is : tcp://localhost:61616"

   private static ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("ApplicationContext.xml");
   private static final Logger logger = LoggerFactory.getLogger(DownloadConsumer.class);

   // Working Code with JMS 2.0
   @JmsListener(destination = "MessageProducerJMSV1")
   public void processBrokerQueues(String message) throws DaoException, JMSException {

      try {

         RequestDao requestDao = (RequestDao) context.getBean("requestDao");

         String receivedStatus = requestDao.getRequestStatus(message);

         //Retrieve Username from the message to include in an email
         String[] parts = message.split("#");
         String userName = parts[1].trim();

         //Retrieve personnelID from the message to include in the webservice calls

         String personnelID = parts[3].trim();

         //Before sending this message, do the check for COMPLETE or ERROR etc
         if (receivedStatus.equals("COMPLETE")) {

            String latestUUID = requestDao.getUUID();

            logger.info("Received UUID in Controller is as follows! ");
            logger.info(latestUUID);

            requestDao.sendMessage(message, latestUUID);
            logger.info("Received status is COMPLETE! ");
            logger.info("Sending email to the user! ");
            String emailMessage = "Dear " + userName + ",<p>Your files are ready. </p><p> Thanks,<br/> Jack/p>";
            String recipientEmail = userName + "@organization.com";

            /*****************************************************\
             // START: EMAIL Related Code
             *******************************************************/

            MimeMessage msg = javaMailSender.createMimeMessage();
            MimeMessageHelper helper = new MimeMessageHelper(msg, true);
            helper.setFrom("[e-mail]");
            helper.setTo(recipientEmail);
            helper.setSubject("Requested Files !");
            helper.setText(emailMessage, true);

            javaMailSender.send(msg);

         } else {
            try {
               createConnection();

               Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

               // Destination represents here our queue 'MessageProducerJMSV1' on the  JMS server
               Destination destination = session.createQueue(subject);

               MessageProducer producer = session.createProducer(destination);

               //Sending message to the queue
               TextMessage toSendMessage = session.createTextMessage(message);

               long delay = 300 * 1000;
               long period = 300 * 1000;

               toSendMessage.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, delay);

               producer.send(toSendMessage);
            } catch (Throwable th) {
               th.printStackTrace();
               
               synchronized (connectionLock) {
                  // if there are any problems close the connection and it will be re-created next time
                  if (connection != null) {
                     connection.close();
                     connection = null;
                  }
               }
            }
         }
      } catch (Throwable th) {
         th.printStackTrace();
      }
   }
   
   private void createConnection() {
      synchronized (connectionLock) {
         if (connection == null) {
            // Getting JMS connection from the server
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
            connection = connectionFactory.createConnection();
         }
      }
   }
}

[Tim Holloway]

If performance is a problem, then adding synchronization will only make it worse. Synchronization will bottleneck all requests. Far better, if possible to use pooled connections with one connection per requester.

Also, why invoke the BeanFactory to get the DAO? Why can't you have it injected? I do.

[Dave Tolls]

Not only what Tim says re: creating a bottleneck, but I'm not convinced that the synching you have there will actually work.
Most of the code using the connection is not synched, so you'll still hit the same issues. It's quite possible for the connection to be closed/changed/operated on, while some other thread is using it.

Just use a connection pool.
It's what they were built for.