Camel, openmq and tomcat: hangup on shutdown?

[cmmo]

Hello everybody,

I wrote a minimal web application which, on application server start,
instances a camel context configuring an activeMQ component (running as
a standalone broker), then starts it, storing it in the servlet context
and configures a route from activeMQ. It then stops the camel context
when the application server stops.

I made it run on tomcat and it apparently works, processing messages
from the activeMQ queue; the relevant versions are: camel 1.5.0,
ActiveMQ 5.2.0, tomcat 6.0.14, jvm 1.6.0 on linux.

The problem happens on tomcat shutdown: one of the activeMQ threads
(called "ActiveMQ Transport") doesn't stop after the camel context is
stopped, which is really not desirable as it prevents tomcat from
shutting down.

Looking more in depth the thread is processing keepalive messages from
activeMQ and keeps processing them "forever" (I tried to wait for some
tens of minutes hoping for a timeout) just as if the activeMQ connection
was not brought down.

As far as I know the only way to stop it is by stopping the activeMQ
broker, in which case the "ActiveMQ Transport" thread in my web
application dies with this NPE:

Exception in thread "ActiveMQ Transport: tcp://localhost/127.0.0.1:61616"
java.lang.NullPointerException
         at
org.apache.activemq.transport.InactivityMonitor.stopMonitorThreads(InactivityMonitor.java:286)
         at
org.apache.activemq.transport.InactivityMonitor.onException(InactivityMonitor.java:243)
         at
org.apache.activemq.transport.TransportSupport.onException(TransportSupport.java:96)
         at
org.apache.activemq.transport.tcp.TcpTransport.run(TcpTransport.java:189)
         at java.lang.Thread.run(Thread.java:619)

As far as I understand after a quick scan of the relevant code: the
thread catches the I/O exception when the TCP socket goes down and then
tries to de-register itself from the inactivity monitor thread list,
which no longer exists. All of this happens in ActiveMQ code and looks
like some shutdown/deinitialization problem either in camel or in
ActiveMQ itself.

I'm attaching the absolutely minimal source for a web application (only
a custom ServletContext performing the above steps) exhibiting this
behaviour. In order to test it you'll need to configure it in the web
application WEB-INF/web.xml, adding:

<listener>
  <listener-class>CustomServletContextListener</listener-class>
</listener>

somewhere inside the <web-app> tag.

Note that this same code, modified to use the JMS camel component
instead of the activeMQ component, shuts down correctly. I'd rather not
do this though, as camel docs clearly state it's not the ideal solution
when talking to activeMQ.

My question is "I'm obviously doing something wrong, but what"?

I know it might be an ActiveMQ problem rather than a camel one but
camel/activemq interactions are not so linear and so I'm asking here first.

Thanks everybody in advance for their answers and suggestions,
Michele


import java.util.Enumeration;
import java.util.Properties;

import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;
import javax.servlet.ServletContext;
import javax.servlet.ServletContextEvent;
import javax.servlet.ServletContextListener;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.camel.component.ActiveMQComponent;
import org.apache.camel.CamelContext;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.jms.JmsComponent;
import org.apache.camel.component.jms.JmsConfiguration;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.log4j.PropertyConfigurator;

public class CustomServletContextListener implements ServletContextListener {
        public CustomServletContextListener() {
        }

        @Override
        public void contextDestroyed(ServletContextEvent evt) {
                final ServletContext servletContext = evt.getServletContext();

                CamelContext camelContext = (CamelContext)servletContext.getAttribute("CamelCtx");
                if (camelContext != null) {
                        servletContext.removeAttribute("CamelCtx");
                        try {
                                camelContext.stop();
                        } catch (Exception e) {
                                throw new IllegalArgumentException("Error stopping the Camel context", e);
                        }
                }
        }

        @SuppressWarnings("unchecked")
        @Override
        public void contextInitialized(ServletContextEvent evt) {
                final ServletContext servletContext = evt.getServletContext();

                Context jndiContext;
                try {
                        jndiContext = new InitialContext();
                } catch (NamingException e) {
                        throw new IllegalArgumentException("Cannot retrieve JNDI initial context", e);
                }

                boolean useJms = false;
               
                if (servletContext.getAttribute("CamelCtx") == null) {
                        CamelContext camelContext = new DefaultCamelContext(jndiContext);
                        if (useJms) {
                                JmsConfiguration cfg = new JmsConfiguration(new ActiveMQConnectionFactory("tcp://localhost:61616"));
                                camelContext.addComponent("activemq", JmsComponent.jmsComponent(cfg));
                        } else {
                                camelContext.addComponent("activemq", ActiveMQComponent.activeMQComponent("tcp://localhost:61616"));
                        }
                        try {
                                camelContext.addRoutes(new RouteBuilder() {
                                        @Override
                                        public void configure() throws Exception {
                                                from("activemq:TestIn").to("mock:null");
                                        }
                                });
                        } catch (Exception e) {
                                throw new IllegalArgumentException("Error configuring Camel routes", e);
                        }

                        try {
                                camelContext.start();
                        } catch (Exception e) {
                                throw new IllegalArgumentException("Error starting the Camel context", e);
                        }
                        servletContext.setAttribute("CamelCtx", camelContext);
                }
        }
}

[Ashwin Karpe]

Hi,

You might want to add some transport options to the client<-->broker connection such as connectionTimeout, soTimeout etc. While I am not sure which option or combination may specifically alleviate this issue, it should address the issue related to keepalives and socket connection.

Check out the following link for relevant options

   http://activemq.apache.org/tcp-transport-reference.html

Hope this helps.

Cheers,

Ashwin...

cmmo wrote:

Hello everybody,

I wrote a minimal web application which, on application server start,
instances a camel context configuring an activeMQ component (running as
a standalone broker), then starts it, storing it in the servlet context
and configures a route from activeMQ. It then stops the camel context
when the application server stops.

I made it run on tomcat and it apparently works, processing messages
from the activeMQ queue; the relevant versions are: camel 1.5.0,
ActiveMQ 5.2.0, tomcat 6.0.14, jvm 1.6.0 on linux.

The problem happens on tomcat shutdown: one of the activeMQ threads
(called "ActiveMQ Transport") doesn't stop after the camel context is
stopped, which is really not desirable as it prevents tomcat from
shutting down.

Looking more in depth the thread is processing keepalive messages from
activeMQ and keeps processing them "forever" (I tried to wait for some
tens of minutes hoping for a timeout) just as if the activeMQ connection
was not brought down.

As far as I know the only way to stop it is by stopping the activeMQ
broker, in which case the "ActiveMQ Transport" thread in my web
application dies with this NPE:

Exception in thread "ActiveMQ Transport: tcp://localhost/127.0.0.1:61616"
java.lang.NullPointerException
         at
org.apache.activemq.transport.InactivityMonitor.stopMonitorThreads(InactivityMonitor.java:286)
         at
org.apache.activemq.transport.InactivityMonitor.onException(InactivityMonitor.java:243)
         at
org.apache.activemq.transport.TransportSupport.onException(TransportSupport.java:96)
         at
org.apache.activemq.transport.tcp.TcpTransport.run(TcpTransport.java:189)
         at java.lang.Thread.run(Thread.java:619)

As far as I understand after a quick scan of the relevant code: the
thread catches the I/O exception when the TCP socket goes down and then
tries to de-register itself from the inactivity monitor thread list,
which no longer exists. All of this happens in ActiveMQ code and looks
like some shutdown/deinitialization problem either in camel or in
ActiveMQ itself.

I'm attaching the absolutely minimal source for a web application (only
a custom ServletContext performing the above steps) exhibiting this
behaviour. In order to test it you'll need to configure it in the web
application WEB-INF/web.xml, adding:

<listener>
  <listener-class>CustomServletContextListener</listener-class>
</listener>

somewhere inside the <web-app> tag.

Note that this same code, modified to use the JMS camel component
instead of the activeMQ component, shuts down correctly. I'd rather not
do this though, as camel docs clearly state it's not the ideal solution
when talking to activeMQ.

My question is "I'm obviously doing something wrong, but what"?

I know it might be an ActiveMQ problem rather than a camel one but
camel/activemq interactions are not so linear and so I'm asking here first.

Thanks everybody in advance for their answers and suggestions,
Michele


import java.util.Enumeration;
import java.util.Properties;

import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;
import javax.servlet.ServletContext;
import javax.servlet.ServletContextEvent;
import javax.servlet.ServletContextListener;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.camel.component.ActiveMQComponent;
import org.apache.camel.CamelContext;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.jms.JmsComponent;
import org.apache.camel.component.jms.JmsConfiguration;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.log4j.PropertyConfigurator;

public class CustomServletContextListener implements ServletContextListener {
        public CustomServletContextListener() {
        }

        @Override
        public void contextDestroyed(ServletContextEvent evt) {
                final ServletContext servletContext = evt.getServletContext();

                CamelContext camelContext = (CamelContext)servletContext.getAttribute("CamelCtx");
                if (camelContext != null) {
                        servletContext.removeAttribute("CamelCtx");
                        try {
                                camelContext.stop();
                        } catch (Exception e) {
                                throw new IllegalArgumentException("Error stopping the Camel context", e);
                        }
                }
        }

        @SuppressWarnings("unchecked")
        @Override
        public void contextInitialized(ServletContextEvent evt) {
                final ServletContext servletContext = evt.getServletContext();

                Context jndiContext;
                try {
                        jndiContext = new InitialContext();
                } catch (NamingException e) {
                        throw new IllegalArgumentException("Cannot retrieve JNDI initial context", e);
                }

                boolean useJms = false;
               
                if (servletContext.getAttribute("CamelCtx") == null) {
                        CamelContext camelContext = new DefaultCamelContext(jndiContext);
                        if (useJms) {
                                JmsConfiguration cfg = new JmsConfiguration(new ActiveMQConnectionFactory("tcp://localhost:61616"));
                                camelContext.addComponent("activemq", JmsComponent.jmsComponent(cfg));
                        } else {
                                camelContext.addComponent("activemq", ActiveMQComponent.activeMQComponent("tcp://localhost:61616"));
                        }
                        try {
                                camelContext.addRoutes(new RouteBuilder() {
                                        @Override
                                        public void configure() throws Exception {
                                                from("activemq:TestIn").to("mock:null");
                                        }
                                });
                        } catch (Exception e) {
                                throw new IllegalArgumentException("Error configuring Camel routes", e);
                        }

                        try {
                                camelContext.start();
                        } catch (Exception e) {
                                throw new IllegalArgumentException("Error starting the Camel context", e);
                        }
                        servletContext.setAttribute("CamelCtx", camelContext);
                }
        }
}

[m.s.]

cmmo wrote:

The problem happens on tomcat shutdown: one of the activeMQ threads
(called "ActiveMQ Transport") doesn't stop after the camel context is
stopped, which is really not desirable as it prevents tomcat from
shutting down.

Did you come up with a solution for this? I encouter the same problem: after removing a route that consumes messages from ActiveMQ from the CamelContext, the ActiveMQ transport thread keeps running.