DZone
Thanks for visiting DZone today,
Edit Profile
  • Manage Email Subscriptions
  • How to Post to DZone
  • Article Submission Guidelines
Sign Out View Profile
  • Post an Article
  • Manage My Drafts
Over 2 million developers have joined DZone.
Log In / Join
Refcards Trend Reports
Events Video Library
Over 2 million developers have joined DZone. Join Today! Thanks for visiting DZone today,
Edit Profile Manage Email Subscriptions Moderation Admin Console How to Post to DZone Article Submission Guidelines
View Profile
Sign Out
Refcards
Trend Reports
Events
View Events Video Library
Zones
Culture and Methodologies Agile Career Development Methodologies Team Management
Data Engineering AI/ML Big Data Data Databases IoT
Software Design and Architecture Cloud Architecture Containers Integration Microservices Performance Security
Coding Frameworks Java JavaScript Languages Tools
Testing, Deployment, and Maintenance Deployment DevOps and CI/CD Maintenance Monitoring and Observability Testing, Tools, and Frameworks
Culture and Methodologies
Agile Career Development Methodologies Team Management
Data Engineering
AI/ML Big Data Data Databases IoT
Software Design and Architecture
Cloud Architecture Containers Integration Microservices Performance Security
Coding
Frameworks Java JavaScript Languages Tools
Testing, Deployment, and Maintenance
Deployment DevOps and CI/CD Maintenance Monitoring and Observability Testing, Tools, and Frameworks

Modern Digital Website Security: Prepare to face any form of malicious web activity and enable your sites to optimally serve your customers.

Containers Trend Report: Explore the current state of containers, containerization strategies, and modernizing architecture.

E-Commerce Development Essentials: Considering starting or working on an e-commerce business? Learn how to create a backend that scales.

Related

  • Everything You Need to Know and Do With Load Balancers
  • Integrating Salesforce With Google BigQuery for Cortex Framework Deployment
  • Managed MQTT Broker Comparison — Console/Dashboard Features
  • Managed MQTT Broker Comparison — Product Packages and Pricing

Trending

  • Data Ingestion for Batch/Near Real-Time Analytics
  • From Sprout to Sequoia: Orchestrating the Symphony of IoT Growth and Cloud Scalability
  • Navigating API Governance: Best Practices for Product Managers
  • Announcing DZone Core 2.0!

A JMS Connection Leak Scenario in Camel

Torsten Mielke user avatar by
Torsten Mielke
·
Aug. 27, 13 · Interview
Like (0)
Save
Tweet
Share
7.8K Views

Join the DZone community and get the full member experience.

Join For Free

I came across this interesting problem today and want to capture it.  Suppose this simple Camel route, consuming from ActiveMQ and logging the message:

<camelcontext id="camelContext" xmlns="http://camel.apache.org/schema/spring">
  <route id="jms-consumer4">
    <from uri="amq:queue:Test" />
    <to uri="log:JMSConsumer?level=INFO&showBody=true" />
  </route>
</camelcontext>
 
<bean class="org.apache.activemq.camel.component.ActiveMQComponent" id="amq">
  <property name="connectionFactory" ref="singleCF" />
  <property name="useSingleConnection" value="true" />
  <property name="usePooledConnection" value="false" />
  <property name="preserveMessageQos" value="true" />
</bean>
 
<bean class="org.springframework.jms.connection.SingleConnectionFactory" id="singleCF">
  <property name="targetConnectionFactory" ref="AMQCF" />
  <property name="reconnectOnException" value="true" />
</bean>
 
<bean class="org.apache.activemq.ActiveMQConnectionFactory" id="AMQCF">
  <property name="userName" value="admin" />
  <property name="password" value="admin" />
  <property name="brokerURL" value="tcp://localhost:61616" />
  <property name="copyMessageOnSend" value="false" />
  <property name="useAsyncSend" value="true" />
</bean>

After deploying this to ServiceMix (JBoss Fuse 6.0 in my test) it works just fine. The problem occurs when stopping this Camel route (via an osgi:stop of the corresponding bundle). The connection into the ActiveMQ broker does not get closed! Drilling into the JMX view of the broker the connection is still registered in JMX. Even worse each restart and stop of the Camel route leaks another connection.

I did a bit of root cause analysis and found:

When the Camel route is stopped it calls into SingleConnectionFactory.destroy(). This cleans up the internally held ActiveMQConnection. At this stage the connection is properly removed from the broker's JMX view, which you will probably only notice when debugging through the code.
However the Spring JMS listener used by the Camel JMS consumer is still running and it detects that the connection is down and tries to transparently reconnect. This calls into SingleConnectionFactory.createConnection again (full stack trace below [1]). The SingleConnectionFactory does happily reopen a new connection into the broker which then remains registered in JMX and open although the route gets stopped.

So how do you resolve this?

Instead of using Springs SingleConnectionFactory I recommend to use ActiveMQs PooledConnectionFactory instead. So the above Camel route configuration becomes
<bean class="org.apache.activemq.camel.component.ActiveMQComponent" id="amq">
  <property name="connectionFactory" ref="pooledCF" />
  <property name="useSingleConnection" value="true" />
  <property name="usePooledConnection" value="false" />
  <property name="preserveMessageQos" value="true" />
</bean>
 
<bean class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop" id="pooledCF" init-method="start">
  <property name="connectionFactory" ref="AMQCF" />
  <property name="maxConnections" value="1" />
</bean>
 
<bean class="org.apache.activemq.ActiveMQConnectionFactory" id="AMQCF">
  <property name="userName" value="admin" />
  <property name="password" value="admin" />
  <property name="brokerURL" value="tcp://localhost:61616" />
  <property name="copyMessageOnSend" value="false" />
  <property name="useAsyncSend" value="true" />
</bean>

When using the ActiveMQs PooledConnectionFactory instead, things behave pretty much the same with one subtle but important difference.

Similar to above, stopping the Camel bundle calls into PooledConnectionFactory.stop(). This internally closes all ActiveMQConnections (only one in this example case but potentially more) which also unregisters the connection from the brokers JMX view. Now, Springs JMS listener used by the Camel JMS consumer is still running and detects the connection closure and tries to transparently reconnect. This calls into PooledConnectionFactory.createConnection(). This implementation however contains the following check:

if (stopped.get()) {
  LOG.debug("PooledConnectionFactory is stopped, skip create new connection.");
  return null;
}

AtomicBoolean stopped will be set to true and no new connection is established!

Springs SingleConnectionFactory does not have this logic. It happily reopens a new connection after it got destroyed. Please note the properties init-method="start" destroy-method="stop" on the PooledConnectionFactory bean definition are important as otherwise you may also leak connections when shutting down your bundles.

[1]
Daemon Thread [Camel (camelContext) thread #21 - JmsConsumer[EwdTest1]]
(Suspended (breakpoint at line 280 in org.springframework.jms.connection.SingleConnectionFactory))
owns: java.lang.Object (id=9254)
owns: java.lang.Object (id=9255)
owns: java.lang.Object (id=9286)
org.springframework.jms.connection.SingleConnectionFactory.initConnection() line: 280
org.springframework.jms.connection.SingleConnectionFactory.createConnection() line: 225
org.apache.camel.component.jms.DefaultJmsMessageListenerContainer(org.springframework.jms.support.JmsAccessor).createConnection() line: 184
org.apache.camel.component.jms.DefaultJmsMessageListenerContainer(org.springframework.jms.listener.AbstractJmsListeningContainer).createSharedConnection() line: 404
org.apache.camel.component.jms.DefaultJmsMessageListenerContainer(org.springframework.jms.listener.AbstractJmsListeningContainer).refreshSharedConnection() line: 389
org.apache.camel.component.jms.DefaultJmsMessageListenerContainer(org.springframework.jms.listener.DefaultMessageListenerContainer).refreshConnectionUntilSuccessful() line: 869
org.apache.camel.component.jms.DefaultJmsMessageListenerContainer(org.springframework.jms.listener.DefaultMessageListenerContainer).recoverAfterListenerSetupFailure() line: 851
org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.run() line: 982
java.util.concurrent.ThreadPoolExecutor$Worker.runTask(java.lang.Runnable) line: 895
java.util.concurrent.ThreadPoolExecutor$Worker.run() line: 918
java.lang.Thread.run() line: 680


Connection (dance)

Published at DZone with permission of Torsten Mielke, DZone MVB. See the original article here.

Opinions expressed by DZone contributors are their own.

Related

  • Everything You Need to Know and Do With Load Balancers
  • Integrating Salesforce With Google BigQuery for Cortex Framework Deployment
  • Managed MQTT Broker Comparison — Console/Dashboard Features
  • Managed MQTT Broker Comparison — Product Packages and Pricing

Comments

Partner Resources

X

ABOUT US

  • About DZone
  • Send feedback
  • Careers
  • Sitemap

ADVERTISE

  • Advertise with DZone

CONTRIBUTE ON DZONE

  • Article Submission Guidelines
  • Become a Contributor
  • Core Program
  • Visit the Writers' Zone

LEGAL

  • Terms of Service
  • Privacy Policy

CONTACT US

  • 3343 Perimeter Hill Drive
  • Suite 100
  • Nashville, TN 37211
  • support@dzone.com

Let's be friends: