Observations on ActiveMQs temp storage.

In ActiveMQ there are 3 areas where messages are stored by the broker: in broker's memory, in persistence store and in temp storage.
All messages get first put into the broker's memory until it fills up. Persistent messages always get additionally saved to the persistence store (KahaDB by default). Non persistent messages are held in memory until broker memory is exhausted. Then, if the default FileCursor is used, they get swapped out to disk (swapping does not happen with VMCursor). The temp storage typically resides in directory data/localhost/tmp_storage.

All three areas are configured in activemq.xml using

<systemUsage>
  <systemUsage>
    <memoryUsage>
      <memoryUsage limit="20 mb"/>
    </memoryUsage>
    <storeUsage>
      <storeUsage limit="1 gb"/>
    </storeUsage>
    <tempUsage>
      <tempUsage limit="100 mb"/>
    </tempUsage>
  </systemUsage>
</systemUsage>

An important advice is to never set the tempUsage limit below the journalFileSize for the temporary storage (32 MB by default but configurable)! So if you don't explicitly configure the journaleFileSize for temp messages in KahaDB, never set the tempUsage limit below 32 MB!

This configuration will most likely cause problems

<tempUsage>
  <tempUsage limit="30 mb"/>
</tempUsage>

Here is the reason why:

The temp storage is only used with non-persistent messages, so typically with topic messages.
In the case of running fast subscribers that keep up with the producers, you will most likely never need to swap messages as all outstanding messages can all be held in memory.
There may be peak times however where the producer rate is higher than the rate for consuming messages. So swapping of messages might occur.

When a subscription realizes that there is no further heap memory to take on new messages (due to whatever configured memory limit), it will start to swap out all messages it holds to temp storage. All messages of a particular subscription get swapped in one go! That process is rather slow and for a larger amount of messages to be swapped might easily take a few mins.

Side Note 1: The dispatching of topic messages to each topic subscription inside the broker is serialized. If a topic has 5 subscribers then a new message gets dispatched to each subscription in a serialized manner. The first subscription that tries to handle a new message might realize that memory is full and starts to swap all of its yet undelivered messages. The swapping is actually done by the FilePendingMessageCursor. After swapping messages to temp storage, the subscription can handle the new message and send it to the external client. There is one FilePendingMessageCursor instance for each subscription. The next subscription is only called once the previous subscription has sent the message and will then normally find enough space in heap memory again to handle the new message in memory. If not, it will start to swap out its messages too to free more heap.

Side Note 2: Because of all messages being swapped in one go by a subscription, it may happen that after the swap has finished the temp storage is above 100% usage. This might occur either when
1) using a memoryUsage > tempUsage and the subscription holds a large amount of messages, but also
2) when configuring a tempUsage limit < 32 MB (see above).

In the case 1) it is possible that the amount of messages held in a subscription is larger than the tempUsage limit. So after swapping all messages of a subscription into temp storage, the tempUsage is > 100%.
In case 2) once swapping starts, the persistence adapter (which is also responsible for writing messages to temp storage) will create a db-1.log file in temp storage and its default file size is 32 MB. Again tempUsage is > 100% after writing this file irrespective of how many messages got swapped to temp storage.
You might see the following log statement in the broker log file when temp storage has filled up after messages got swapped to disk.

INFO TopicSubscription - TopicSubscription: consumer=ID:nbwfhtmielke-1380-1298467525966-2:1:1:1, destinations=1, dispatched=99, delivered=9622, matched=231, discarded=0: Pending message cursor [org.apache.activemq.broker.region.cursors.FilePendingMessageCursor@3c0737] is full, temp usage (129%) or memory usage (0%) limit reached, blocking message add() pending the release of resources.

With producer flow control enabled, any producers will be put on hold until the swapping of messages has finished (and additional heap memory is available) and also until the tempUsage < 100%. Subscribers still get messages dispatched while tempUsage > 100%, only producers are stopped. Any consumed messages free additional space from temp storage. Once the temp usage is < 100% producers get resumed.

As messages get consumed the data files get deleted from temp storage. All except for the first file which will be reused in case any of new messages that need to be swapped.

And there is the problem: The first data file never gets deleted. Its default size is 32 MB and hence still above the configured tempUsage limit (30 MB in our example). As the file never gets deleted, tempUsage never goes < 100% and therefore the producer is flow controlled forever and never gets resumed.
From an external viewpoint it seems as if either the broker or the producer is hung. The subscribers have consumed all message but producers don't send any further messages.

Some more useful notes on broker memory configuration can be found here.

Lesson learned:

• Don't configure a tempUsage limit < default size of the journal file (32 MB).


• Swapping out messages to temp storage can be rather slow. If you already using producer flow control with topic messages, the VM cursor might be an alternative. It won't swap messages to disk.

Load tests with ActiveMQ

Not sure everyone knows it but ActiveMQ has a maven plug-in that can be easily used to run load tests. Its name is maven-activemq-perf-plugin. Full documentation is available here.
As I often need to run test using a specific broker configuration or having to test a specific broker feature, this plug-in has helped me a few times already. It is also highly useful for trouble shooting as the many configuration options allow you to simulate certain broker usage patterns.

You have many options for setting up the load test like the number of producers and consumers, the message size, acknowledge mode, using Queues or Topics, whether to use JMS transactions and many more.
It also includes samplers that measure your performance; you get a nice summary written at the end of the test run.


In order to use the test suite, simply add it the plug-in to your pom. An example is given here.

I highly recommend it to anyone who wants to quickly run some load tests and measure broker throughput.

How to change the ActiveMQ broker configuration when running inside ServiceMix?

Update: This post does not apply to Fuse ESB Enterprise 7.x and higher. It only applies to Apache ServiceMix and Fuse ESB up to versions 4.x.
 
I mistakenly assumed that I can simply update the embedded ActiveMQ broker configuration at $KARAF_HOME/etc/activemq-broker.xml and restart SMX for these changes to take effect.

That is not the case.

ServiceMix monitors the etc/ folder via an OSGI management agent. It scans the etc/ folder, installs and starts a bundle when it is first placed there. The etc/config.properties file configures this agent:


felix.fileinstall.dir = ${karaf.base}/etc
felix.fileinstall.filter = .*\\.cfg
felix.fileinstall.poll = 1000
felix.fileinstall.noInitialDelay = true


So when ServiceMix starts up the first time the agent will read the activemq-broker.xml file from etc/, wrap it as an OSGI bundle and deploy it into the bundle cache. Changes to etc/activemq-broker.xml thereafter will not cause the bundle to be redeployed automatically, not even after a restart of ServiceMix. This is the actually the same behavior as deploying a bundle using, e.g. osgi:install mvn:… from your local Maven repository. Updating the bundle in your Maven repo does not cause ServiceMix to redeploy it automatically for you (which is certainly good).

So after changing the configuration of etc/activemq-broker.xml, make sure to update the ActiveMQ broker bundle:


karaf@root> list -l | grep activemq-broker.xml
[ 57] [Active ] [Created ] [ ] [60] blueprint:file:etc/activemq-broker.xml
karaf@root>update 57


It will stop the embedded broker, reload the configuration as an OSGI bundle and restart the broker with the new configuration in effect.

Sync your machine clocks!

When running ActiveMQ with producer and consumers spread across multiple machines, make sure to have the clocks synced on these machines!
Otherwise there might be interesting side effects when using JMS expiration times.

1) Consider the following simple scenario:
A broker running on host A with the local time 1.35 pm.
Secondly a producer/consumer pair running on host B with the local time of 1.30 pm.

In summary:
Broker time: 1.35 pm
JMS client time: 1.30 pm


The producer sends a message with a JMSExpiration time of 2 mins at 1.30 pm sharp. So the message expires at 1.32 pm. The broker receives the message, checks the expiration time and realizes the message is already expired. So it gets moved to DLQ immediately. The consumer will not get the message! This might be particularly surprising if the consumer is on the same machine as the producer and you start to wonder where your message is.
Note: The JMSExpiration time that is set on the message does not contain the value 2mins, but the actual time in future when the message expires (represented as a long). This value is computed using the local time of the message producer and compared against the local time of the broker before being put on the queue.

2) Now let's consider the opposite example:
In summary
Broker time: 1.30 pm
JMS client time: 1.35 pm

The broker's local time is 1.30 pm and the producers/consumers local time is 1.35 pm.
The Producer again sends a message with a JMSExpiration of 2 mins. It is received by the broker at 1.30 with an expiration time at 1.37pm. The message gets put onto the queue, from where the consumer can grab it. So all is fine in this scenario.

It is therefore highly suggested that the local times between all parties that participate in messaging are more or less synchronized. One simple option is to configure for NTP synchronization on each machine. There are many NTP tools available for all major operating systems.

If for whatever reason you cannot synchronize the times between machines (e.g. broker running externally), then I suggest to use a JMSExpiration time that include the delta of the time difference between machines. E.g. if the delta is known to be 5 mins, then perhaps set the JMSExpiration time to 5+ mins (adding enough time for message processing and delivery).
On the other hand if message expiration is an important requirement in your application, you should really try to synchronize times between all involved machines.

Part 2

This brings me to the second part of this post, the ActiveMQ TimeStampingBrokerPlugin.

From its documentation:

"This can be useful when the clocks on client machines are known to not be correct and you can only trust the time set on the broker machines."

1) Let's revisit the first scenario again:
Broker time: 1.35 pm
JMS client time: 1.30 pm

The plug-in can help you in this case. Because the plug-in will not only set the JMS message timestamp to the current time at the broker but also recalculate the resulting JMS expiration time again based on the broker's local time. Thus the message will not be marked as expired when it is handled by the broker. It is therefore put onto the queue from where the consumer can grab it. So the use of the TimeStampingBrokerPlugin can help to resolve the problem of scenario 1).


2) Now let's consider the second scenario again:
Broker time: 1.30 pm
JMS client time: 1.35 pm

The JMS producer sends the message at 1.35 pm local time with an expiration time set to 1.37 pm.
The TimeStampingBrokerPlugin resets the expiration time to 1.32 pm (2 mins based on the brokers local time). The message is put onto the queue.
The consumer that is connected has a local time of 1.35 pm. It will not grab the message!!
Why? Because from this consumer's point of view the message has already expired.

Such situation will generally be difficult to understand when looking at the system using either the ActiveMQ web console or JMX console. There is a message on the queue and there is a consumer connected but the message is not consumed!
You might not immediately think about JMS expiration times and different machine times. You will more likely start to think the attached consumer is hung or there is a bug in ActiveMQ.
If you configure for ActiveMQ debug logging in the consumer, you will notice that the consumer actually gets the message from the queue but it will discard it due to its expiration time. Under debug logging the following is printed:

ActiveMQMessageConsumer DEBUG ID:nbwfhtmielke-4668-1294676704879-2:0:1:1 received expired message: MessageDispatch {commandId = 0, … expiration = 1294675812454,
timestamp = 1294674812454, …}

There is a possible solution though, that is to use the plug-in configuration property futureOnly="true". If set to true the plug-in will not set the new expiration time on the message if it is lower than the original expiration time. It will therefore never reset the expiration time to a lower value. Instead the original expiration time gets preserved. That way the remote consumer will grab the message.


Note: You could also run into this problem with a camel-jms route using INOUT message exchange pattern and connecting to an external broker that has this plug-in configured.
camel-jms uses requestTimeout=20 secs by default. That generates a JMSExpiration message header with an expiration time of 20 secs. If the broker's local time is only 30 seconds (or even less) behind the local time of the JMS consumer, the same issue of the message not getting consumed might occur.

Conclusion: If somehow possible, sync the machine clocks on all machines that are involved in the message exchange. If that is not possible, check the time differences and recalculate your JMS expiration times.

Configure client side CXF bus in OSGi bundle

Perhaps it was only me who did not get this right in the first place, but here is what I learned last week.

Inside the same Spring configuration file I configured a CXF bus instance for client side failover and a plain Java bean that was going to use this CXF bus instance for making an external Web Services invocation. This configuration was to be deployed into Apache ServiceMix 4.3 as an OSGI bundle.
Here is the straight-forward Spring configuration:

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans" >

<!-- Via these imports a CXF bus instance will be made available -->
<import resource="classpath:META-INF/cxf/cxf.xml" />
<import resource="classpath:META-INF/cxf/cxf-extension-soap.xml" />
<import resource="classpath:META-INF/cxf/cxf-extension-http.xml" />

<!-- Instantiates my Java bean -->
<bean id="testBean" class="org.tmielke.cxf.failovertest.TestBean"/>


<!-- CXF bus failover configuration: -->

<!-- List of alternative addresses -->
<util:list id="addressList">
  <value>http://localhost:9000/MyService </value>
  <value>http://localhost:9001/MyService </value>
</util:list>

<!-- CXF failover strategy to use above address list -->
<bean id="RandomAddresses" 
  class="org.apache.cxf.clustering.RandomStrategy">
  <property name="alternateAddresses">
    <ref bean="addressList"/>
  </property>
</bean>
 
<!-- CXF bus to use failover strategy -->
<jaxws:client name="{http://com.fusesource/MyService}MyServiceSoap" 
  createdFromAPI="false">
  <jaxws:features>
    <clustering:failover>
      <clustering:strategy>
        <ref bean="RandomAddresses"/>
      </clustering:strategy>
    </clustering:failover>
  </jaxws:features>
</jaxws:client>

</beans>

This example configures a JAX-WS client configuration for failover using a random strategy for selecting a failover server. It uses the bus instance that was made available by the cxf imports.
It also instantiates a plain Java bean called testBean.
In this Java bean I simply want to use JAX-WS APIs to make an invocation to an external Web Service and I want this failover configuration to be in effect.

The JAX-WS client code reads similar to this:

package org.tmielke.cxf.failovertest;
import com.fusesource.test.MyService;
import com.fusesource.test.MyServiceSoap;

public class TestBean {

  public void goForIt() {
    MyService service = new MyService();
    MyServiceSoap proxy = service.getMyServiceSoap();
    proxy.callWhateverBusinessMethod();
  }
}

My assumption was that because I configured both the Java bean and the CXF bus instance inside the same Spring configuration file, that my Java bean use this pre-configured bus instance when making an outgoing JAX-WS invocation.

However, that is not the case!
The problem is that when deploying into OSGi, there is a particular thread used at deployment time that parses the Spring configuration files. Then at runtime another thread is used for executing the Java bean and the JAX-WS code.
The SpringDeployer thread at deployment time creates a CXF bus instance and configures it for failover according to my Spring configuration. However this CXF bus instance is only bound to the thread context of this SpringDeployer thread.

At runtime when the above JAX-WS code gets executed, a different thread will run this code and that thread will have a different CXF thread context assigned. This CXF thread context is not connected to the CXF bus instance created at deployment time, but to a default CXF bus instance that has not got any additional configuration. Hence no failover will happen in this case!

So wondering about the solution?

You may guess it already; a simple solution is to inject the TestBean with the CXF bus instance created at deployment time.

<bean id="testBean" class="org.tmielke.cxf.failovertest.TestBean">
  <property name="bus" ref="cxf"/>
</bean>

This requires the Java bean to expose a setter method for the CXF bus instance:

package org.tmielke.cxf.failovertest;

import com.fusesource.test.MyService;
import com.fusesource.test.MyServiceSoap;

import org.apache.cxf.Bus;
import org.apache.cxf.BusFactory;

public class TestBean {

  //store the configured CXF bus internally
  private Bus bus = null;
  
  public void setBus(Bus bus) {
    this.bus = bus;
  }

  public Bus getBus() {
    return bus;
  }

  public void goForIt() {
    BusFactory.setThreadDefaultBus(bus);
    MyService service = new MyService();
    MyServiceSoap proxy = service.getMyServiceSoap();
    proxy.callWhateverBusinessMethod();
  }
}

Before using any JAX-WS APIs in my Java bean I need to call BusFactory.setThreadDefaultBus() to assign my preconfigured bus instance to the current thread context.

Now I am ready to make the external Web Service invocation with the failover configuration being in effect.


Note, without explicitly setting the CXF bus instance to be used, I would use the instance that gets returned from BusFactory.getDefaultBus(). BusFactory.getDefaultBus() is called by the CXF JAX-WS implementation to set the bus on the CXF server or client.
BusFactory.getDefaultBus() basically returns a default bus instance without any failover configuration, unless I explicitly assign a different bus instance using BusFactory.setThreadDefaultBus().

Rebooting ...

This blog is finally going to be revitalized!

After an entire year I am finally back to work on Open Source again. In November this year I have joined a company called FuseSource as a Senior Engineer and am really excited about the new role!
FuseSource is owned by Progress Software and offers enterprise subscriptions for the Apache Open Source products ActiveMQ, ServiceMix, Camel and CXF. Wanna know what an enterprise subscription consists of? Check it out here. At FuseSource we certify, package and distribute the above Apache projects, backed by enterprise-class professional services to support all phases of development and deployment of these products. A large community has grown fairly quickly around our offerings.

If you're using any of the above products and their uptime and availability becomes an important factor in your business, you then might want to consider getting professional support.
Besides, come along and join our fast growing community!

So watch out for more technical postings on this blog coming up soon…

Using Spring JMS Template for sending messages to ActiveMQ

Something that took me a long way to really understand. And I only got there by debugging through the low level Spring and ActiveMQ code.
There are various articles out there that warn you about using JmsTemplate for producing messages outside any container. A runtime container will usually pool your JMS resources (connection, session and producer), but when using JmsTemplate in a standalone application, no pooling is there per se. The JmsTemplate is certainly not responsible for pooling resources.

I have seen a number of applications that use JmsTemplate outside of any applicaton server container for producing messages. It seems to be more popular than using the plain JMS APIs. No wonder as Spring JMS provides such a nice layer of abstraction that shields all the low level JMS code, just like this snippet:

ConnectionFactory cf2 = new ConnectionFactory("tcp://localhost:61616");

JmsTemplate template = new JmsTemplate(cf2);

template.send("MyQueue", new MessageCreator() {

  public Message createMessage(Session session)

  throws JMSException {

    TextMessage tm = session.createTextMessage();

    tm.setText("A new message");

    return tm;

  }

});

However something that is easily forgotten is that this code will create a new JMS connection, consumer and producer for each call to JmsTemplate.send(). Creating these JMS objects are expensive operations and according to the JMS spec these resources are meant to be re-used and not re-created for every message. Performance is going to suffer dramatically when they are not re-used.
That’s pretty clear to most developers but very easily forgotten.

So what is really needed in order to pool all these JMS resources when sending messages with JmsTemplate? I personally was confused whether I need to configure some Jencks container in order to have the JMS resources pooled or could I use plain ActiveMQ connection factories and how does the configuration need to look like? The various documentations I found only added to that confusion.

The answer is pretty simple: All that is needed is to use the org.apache.activemq.pool.PooledConnectionFactory. It will serve JMS connections from its internal pool and wrap them in a PooledConnection instance. The purpose of the PooledConnection is to also pool any JMS sessions and producers that get created. The pool can be configured of course.
For using the JmsTemplate there is no need to configure any containers like Jencks and others. Also see http://activemq.apache.org/jmstemplate-gotchas.html for some additional information on this topic.
The following simple and complete Spring configuration is enough:

<?xml version="1.0" encoding="UTF-8"?>

<beans xmlns="http://www.springframework.org/schema/beans"

  xmlns:tx="http://www.springframework.org/schema/tx"

  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

  xsi:schemaLocation="http://www.springframework.org/schema/beans 

  http://www.springframework.org/schema/beans/spring-beans-2.0.xsd

  http://www.springframework.org/schema/tx 

  http://www.springframework.org/schema/tx/spring-tx-2.0.xsd">

	

  <bean id="connectionFactory" 

    class="org.apache.activemq.pool.PooledConnectionFactory">

    <constructor-arg value="failover:(tcp://localhost:61616)"/>

  </bean>



  <bean id="jmsTemplate" 

    class="org.springframework.jms.core.JmsTemplate">

    <constructor-arg ref="connectionFactory"/>

    <property name="sessionTransacted" value="false"/>

    <property name="receiveTimeout" value="5000"/>   

  </bean>

</beans>

It pre-configures the JmsTemplate to use the ActiveMQ PooledConnectionFactory so it can be used straight away in a Java program:

package org.apache.activemq.test.jmstemplate;



import javax.jms.JMSException;

import javax.jms.Message;

import javax.jms.Session;

import javax.jms.TextMessage;



import org.springframework.context.ApplicationContext;

import org.springframework.context.support.ClassPathXmlApplicationContext;

import org.springframework.jms.core.JmsTemplate;

import org.springframework.jms.core.MessageCreator;





public class JmsTemplateTest {



  public static void main(String[] args) throws Exception {



    ApplicationContext ctx = new ClassPathXmlApplicationContext(

      "JmsTemplateTest-context.xml");

    JmsTemplate template = (JmsTemplate) ctx.getBean("jmsTemplate");

        

    for(int i=0; i<10; i++) {

      template.send("MyQueue", new MessageCreator() {

        public Message createMessage(Session session)

          throws JMSException {

          TextMessage tm = session.createTextMessage();

          tm.setText("This is a test message");

          return tm;

        }

      });

    }

    System.exit(1);

  }

}

A Maven based test case is provided. It can be checked out using

svn checkout http://abloggerscode.googlecode.com/svn/trunk/JmsTemplateDemo

Once checked out, follow the instructions in README.txt.

Be careful when using Springs SingleConnectionFactory. It does re-use the underlying JMS connection but it does not re-use the JMS session and producer when configured to internally work with the ActiveMQConnectionFactory!

Does ActiveMQ 5.3 broker get hung if there are millions of messages on a queue and direct JDBC is used?

That was an interesting ActiveMQ issue I had to work on last week.
We used ActiveMQ 5.3.0.3-fuse with direct JDBC persistence to a MySQL database.

And noticed that when we had 2 million persistent messages on queue A, we could not even process any messages on an empty queue B. It seemed the broker got hung but then it did react to any commands from the web or JMX console.

When looking into the cause of this more deeply we found this:

In ActiveMQ there is a cleanup task thread that periodically checks for expired or acknowledged messages. In case of using a direct JDBC persistence, this task runs the following SQL command against the JDBC database table where the JMS messages are stored:

DELETE FROM ACTIVEMQ_MSGS WHERE ( EXPIRATION<>0 AND EXPIRATIONOR ID <= 
( SELECT min(ACTIVEMQ_ACKS.LAST_ACKED_ID) FROM ACTIVEMQ_ACKS WHERE 
ACTIVEMQ_ACKS.CONTAINER=ACTIVEMQ_MSGS.CONTAINER)

Depending on the JDBC database used, that SQL statement might lock the entire database table during the run of that statement. This is certainly the case when using MySQL but not when using Apache Derby. Other JDBC databases might lock the entire table as well.

During that time no other persistent message (no matter for what queue) can be processed, as all JMS messages for all queues are stored in the same database table called activemq_msgs. For a database table with millions of messages, this might take several mins or more (I had to wait around 15 mins for that statement to complete).
Also, the cleanup task is scheduled to run at a fixed rate (every 2 mins or so), so once it completed it is already late on schedule and therefore gets kicked off again straight away, not leaving much time for other threads to insert new messages into the db. So the threads managing the other queues starve and might perhaps never get the processing time to process their messages.

This behavior can be confirmed by adding the following logging configuration

log4j.logger.org.apache.activemq.store.jdbc=DEBUG

It will log the above SQL statement being executed for a long time and being called repeatedly as soon as it has finished.
Also when attaching jconsole to the broker and capturing a stack trace of the thread that is to process new messages, it should be waiting for a long time on the JDBC database driver to complete the SQL insert statement.

Name: QueueThread:queue://JMSLoadTest.queue.0
 State: RUNNABLE
 Total blocked: 1  Total waited: 0

 Stack trace:
 java.net.SocketInputStream.socketRead0(Native Method)
 java.net.SocketInputStream.read(SocketInputStream.java:129)
 com.mysql.jdbc.util.ReadAheadInputStream.fill(ReadAheadInputStream.java:113)
 com.mysql.jdbc.util.ReadAheadInputStream.readFromUnderlyingStreamIfNecessary(ReadAheadInputStream.java:160)
 com.mysql.jdbc.util.ReadAheadInputStream.read(ReadAheadInputStream.java:188)
 com.mysql.jdbc.MysqlIO.readFully(MysqlIO.java:2428)
 com.mysql.jdbc.MysqlIO.reuseAndReadPacket(MysqlIO.java:2882)
 com.mysql.jdbc.MysqlIO.reuseAndReadPacket(MysqlIO.java:2871)
 com.mysql.jdbc.MysqlIO.checkErrorPacket(MysqlIO.java:3414)
 com.mysql.jdbc.MysqlIO.sendCommand(MysqlIO.java:1936)
 com.mysql.jdbc.MysqlIO.sqlQueryDirect(MysqlIO.java:2060)
 com.mysql.jdbc.ConnectionImpl.execSQL(ConnectionImpl.java:2542)
 com.mysql.jdbc.PreparedStatement.executeInternal(PreparedStatement.java:1734)
 com.mysql.jdbc.PreparedStatement.executeQuery(PreparedStatement.java:1876)
 org.apache.commons.dbcp.DelegatingPreparedStatement.executeQuery(DelegatingPreparedStatement.java:91)
 org.apache.commons.dbcp.DelegatingPreparedStatement.executeQuery(DelegatingPreparedStatement.java:91)
 org.apache.activemq.store.jdbc.adapter.DefaultJDBCAdapter.doRecoverNextMessages(DefaultJDBCAdapter.java:709)
 org.apache.activemq.store.jdbc.JDBCMessageStore.recoverNextMessages(JDBCMessageStore.java:230)
 org.apache.activemq.store.ProxyMessageStore.recoverNextMessages(ProxyMessageStore.java:83)
 org.apache.activemq.broker.region.cursors.QueueStorePrefetch.doFillBatch(QueueStorePrefetch.java:75)
 org.apache.activemq.broker.region.cursors.AbstractStoreCursor.fillBatch(AbstractStoreCursor.java:210)
 org.apache.activemq.broker.region.cursors.AbstractStoreCursor.hasNext(AbstractStoreCursor.java:119)
 org.apache.activemq.broker.region.cursors.StoreQueueCursor.hasNext(StoreQueueCursor.java:131)
 org.apache.activemq.broker.region.Queue.doPageIn(Queue.java:1243)
 org.apache.activemq.broker.region.Queue.pageInMessages(Queue.java:1378)
 org.apache.activemq.broker.region.Queue.iterate(Queue.java:1086)
 org.apache.activemq.thread.DeterministicTaskRunner.runTask(DeterministicTaskRunner.java:84)
 org.apache.activemq.thread.DeterministicTaskRunner$1.run(DeterministicTaskRunner.java:41)
 java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:650)
 java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:675)
 java.lang.Thread.run(Thread.java:595)

The broker itself is not hung but the cleanup task locks the entire JDBC database. If the broker is left running for a longer time, it will process new messages, but only a few within a longer time frame.
I raised MB-558 (mirrored in AMQ-2470) for this problem and Gary Tully fixed it by changing the clean up task from a running at a fixed rate to running at a fixed delay.

Understanding Authentication and Authorization in ServiceMix

I spent a good few hours lately trying to understand how authentication and authorization work inside ServiceMix 3. It is this gathered knowledge that I want to share in this post. For those in a hurry, there's an executive summary at the end. For this article I assume you know the basic concepts of JAAS. Some reference material on JAAS can be found here and here.

Such complex topic is easier to understand if it is based on a valid use-case that others can reproduce easily. A pretty good candidate is the cxf-ws-security demo in ServiceMix 3.2. Anyone not familiar with that demo might want to build and run it prior to reading on.
In addition I encourage you to debug through the ServiceMix source code while you read through this article. Simply set the SERVICEMIX_DEBUG=1 environment variable and attach a Java debugger to ServiceMix. You can then place breakpoints into the relevant source code that I discuss below and walk through the call stack and source code as you read along.

I will not try to discuss all of the security aspects in ServiceMix 3 here as such scope is far too broad. Rather this article is based on running the cxf-ws-security demo and aims to explain how an incoming SOAP/HTTP request gets authenticated and authorized in ServiceMix 3.

Alright, the stage is set, let's start.
The demo's use-case is
External CXF Java client -> CXF-BC consumer -> CXF-SE component

as shown in the demo's README.txt. With regards to security we want to make sure that the client gets authenticated based on a username/password combination before verifying if the client has permissions to call the service (authorization). Sounds simple enough but it involves a good number of components as well we see next.

As documented in the cxf-ws-security demo, an external SOAP client sends a SOAP/HTTP request to the CXF-BC consumer component deployed in ServiceMix. The SOAP request includes a large WSSE Security header carrying a WS-Security UsernameToken profile among an XML-Signature and some XML-Encrypted content. I will not focus on XML-Signature and XML-Encryption here (some information on this regards was posted previously). For this discussion the UsernameToken profile is of most interest and is included in the SOAP request as follows (omitting some details for clarity):

<soap:Envelope xmlns:soap="..." >
  <soap:Header>
    ...
    <wsse:Security>
      <wsse:UsernameToken>
        <wsse:Username>alice</wsse:Username>
        <wsse:Password>password</wsse:Password>
      </wsse:UsernameToken>
      ...

It is this username/password combination in the SOAP security header that will be used for authentication inside ServiceMix.
So the request first hits the jetty server in the servicemix-cxf-bc component from where it gets passed into the CXF interceptor stack. There are a good couple of CXF interceptors configured by default. In our demo the list of CXF interceptors to be executed is as follows:

DEBUG - PhaseInterceptorChain:          
  receive [AttachmentInInterceptor, LoggingInInterceptor]
  post-stream [StaxInInterceptor]
  read [ReadHeadersInterceptor]
  pre-protocol [MustUnderstandInterceptor, SAAJInInterceptor, WSS4JInInterceptor, 
                JbiJAASInterceptor]
  unmarshal [JbiOperationInterceptor]
  pre-invoke [JbiInWsdl1Interceptor, JbiInInterceptor]
  invoke [JbiInvokerInterceptor, UltimateReceiverMustUnderstandInterceptor]
  post-invoke [JbiPostInvokerInterceptor, OutgoingChainInterceptor]

Those interceptors in italics are of primary interest for this discussion.
In the configuration of the demo's ws-security-cxfbc-su component (in ws-security-cxfbc-su /src/main/resources/xbean.xml) we explicitly added and configured the org.apache.cxf.ws.security.wss4j.WSS4JInInterceptor that will perform the WS-Security functions. It is the first security relevant interceptor that we want to look at closer.

org.apache.cxf.ws.security.wss4j.WSS4JInInterceptor.handleMessage()

This interceptor handles the entire SOAP WS-Security header. In our example it needs to decrypt all the encrypted parts of the SOAP request, perform the XML-Signature check and extract the username/password information from the WS-Security header. What functions to perform is specified in xbean.xml, where this interceptor is configured (see the "action" list):

<bean class="org.apache.cxf.ws.security.wss4j.WSS4JInInterceptor"
    id="WSS4J">
    <constructor-arg>
        <map>
          <entry key="action" value="Timestamp Signature Encrypt UsernameToken"/>
            <entry key="..." .../>
            ...
        </map>
      </constructor-arg>
    </bean>

The name tells it already, this interceptor uses Apache WSS4J to perform these WS-Security functions. This line of code inside CXF WSS4JInInterceptor.handleMessage() method calls into WSS4J:

Vector wsResult = getSecurityEngine().processSecurityHeader(
  doc.getSOAPPart(), 
  actor, 
  cbHandler, 
  reqData.getSigCrypto(), 
  reqData.getDecCrypto()
);

I will not cover the Apache WSS4J specifics here but want to mention that the WSS4JInInterceptor configuration includes a password callback implementation that is used to obtain password for decryption and UsernameToken verification. This is also specified in the xbean.xml of the servicemix-cxf-bc-su component:

<entry key="passwordCallbackClass" 
   value="org.apache.servicemix.samples.cxf_ws_security.KeystorePasswordCallback"/>

See the Apache WSS4J documentation for more detailed information on WSS4J.

The outcome of each of the WS-Security function performed by WSS4J is stored in a java.util.Vector of org.apache.ws.security.WSSecurityEngineResult objects called wsResult that is then added as a property to the SoapMessage using the key WSHandlerConstants.RECV_RESULTS. This vector carries all the results from processing the WS-Security header, such as the principal, SAML token, X.509 certificates, etc.

Other interceptors that get invoked later will require these results for their own processing.

One of the vector elements will contain the result of parsing the WS-Security UsernameToken profile in form of a org.apache.ws.security.WSUsernameTokenPrincipal object instance. This hosts the username and password information that was received with the SOAP request.

That is roughly what the WSS4JInInterceptor does with our SOAP request.
The next CXF interceptor to be invoked is:

org.apache.servicemix.cxfbc.interceptors.JbiJAASInterceptor.handleMessage()
This one is invoked right after the WSSJ4InInterceptor and is rather complex. I will try my best to explain it in most simple terms.
It first of all extracts the WSHandlerConstants.RECV_RESULTS vector that was set by the previous WSS4JInInterceptor

// in JbiJAASInterceptor.handleMessage()
List<Object> results = (Vector<Object>)message.get(WSHandlerConstants.RECV_RESULTS);

and next checks if this Vector contains any org.apache.ws.security.WSUsernameTokenPrincipal object:

for (Iterator it = hr.getResults().iterator(); it.hasNext();) {
  WSSecurityEngineResult er = (WSSecurityEngineResult) it.next();
  if (er != null && er.getPrincipal() instanceof     
      WSUsernameTokenPrincipal){
    WSUsernameTokenPrincipal p = 
      (WSUsernameTokenPrincipal)er.getPrincipal();
    subject.getPrincipals().add(p);
                           
    this.authenticationService.authenticate(subject, domain, p.getName(), p.getPassword());
  }
}

If yes, then the principal gets added to the javax.security.auth.Subject instance and the authentication service gets invoked. It is this security Subject instance that will carry all credentials information after a successful authentication!

org.apache.servicemix.jbi.security.auth.impl.JAASAuthenticationService.authenticate()
JAAS will be used now to perform the authentication.
At first a new JAAS javax.security.auth.login.LoginContext gets created while also registering a JAAS callback handler:

//in method authenticate()
LoginContext loginContext = new LoginContext(domain, subject, new CallbackHandler() {
  public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException {
    for (int i = 0; i < callbacks.length; i++) {
      if (callbacks[i] instanceof NameCallback) {
        ((NameCallback) callbacks[i]).setName(user);
      } else if (callbacks[i] instanceof PasswordCallback && credentials instanceof String) {
       ((PasswordCallback) callbacks[i]).setPassword(((String) credentials).toCharArray());
      } else if (callbacks[i] instanceof CertificateCallback && credentials instanceof X509Certificate) {
        ((CertificateCallback) callbacks[i]).setCertificate((X509Certificate) credentials);
      } else {
        throw new UnsupportedCallbackException(callbacks[i]);
      }
    }
  }
});

This callback handler that we pass into the LoginContext constructor is able to deal with different types of callbacks (see the handle() method) and will be invoked later in order to take the username and password of the incoming request and place it into the callback object.
Once the LoginContext got instantiated, we can perform the authentication by calling

loginContext.login();

This API call internally invokes the configured JAAS login modules to perform their respective types of authentication (username/password based or certificate based, etc). If the call to login() returns without throwing an exception, the overall authentication succeeded. It is therefore the login module implementation that really decides whether a client is authenticated or not.

How are the login modules configured resolved and configured at runtime? Well, the domain argument of the LoginContext constructor specifies the index into a configuration that determines which LoginModules to use for the authentication. In our case the domain value will be "servicemix-domain", which matches the domain defined in conf/login.properties:

/* conf/login.properties */
servicemix-domain {
    org.apache.servicemix.jbi.security.login.PropertiesLoginModule 
        sufficient
        org.apache.servicemix.security.properties.user="users-passwords.properties"
        org.apache.servicemix.security.properties.group="groups.properties";    

    org.apache.servicemix.jbi.security.login.CertificatesLoginModule 
        sufficient
        org.apache.servicemix.security.certificates.user="users-credentials.properties"
        org.apache.servicemix.security.certificates.group="groups.properties";   
};

By default this registers two login modules which will be invoked in order until the call to login() succeeds. Actually the JAAS javax.security.auth.login.LoginModule will invoke initialize(), login() and commit() on the configured login modules in this order. Each login module class gets configured for a bunch of config files (we'll explore that later).

So the call to loginContext.login() internally invokes

• PropertiesLoginModule.initialize()


• PropertiesLoginModule.login()


• PropertiesLoginModule.commit()

in this order. If the call to any of these methods raises an exception, the CertificatesLoginModule will be tried. If the CertificatesLoginModule also raises an exception, then the authentication fails.
The various classes used in JAASAuthenticationService.authenticate() method (LoginContext, LoginModule, CallbackHandler, etc) are pure JAAS terminology and not ServiceMix specific. If you are not familiar with these, check the JAAS documentation.

Time to examine the org.apache.servicemix.jbi.security.login.PropertiesLoginModule next and see how it performs the authentication.


org.apache.servicemix.jbi.security.login.PropertiesLoginModule
This JAAS login module is used for username/password based login and is configured for two property files in conf/login.properties. The file users-passwords.properties (to be found in your ServiceMix conf/ directory) stores usernames and their passwords. The other properties file groups.properties maps users to their roles. If authentication succeeds, each authenticated user gets one or more roles assigned, according to this properties file. The authorization decision that is performed later will be done on the user's role, not the user's name (role based access control)!
PropertiesLoginModule.initialize() does not do too much, it only resolves the two property file names from the login module configuration in login.properties.
In PropertiesLoginModule.login() these files get loaded into memory. Notice that any changes to either users-passwords.properties or groups.properties will be reloaded when dealing with the next invocation! So you can adjust the user and role mapping at runtime.
The next step is to instantiate a javax.security.auth.callback.Callback array containing a NameCallback and a PasswordCallback.

// in method PropertiesLoginModule.login()
Callback[] callbacks = new Callback[2];
callbacks[0] = new NameCallback("Username: ");
callbacks[1] = new PasswordCallback("Password: ", false);

These callbacks are then passed into the callback handler that was created with the LoginContext earlier in JAASAuthenticationService.authenticate() (see previous sample code above):

//in method PropertiesLoginModule.login()
callbackHandler.handle(callbacks);

The callback handler now iterates through all the callbacks (in our case two, namely the NameCallback and PasswordCallback) and checks of what type they are. Check the code that I quoted above already:

//in method JAASAuthenticationService.authenticate()
LoginContext loginContext = new LoginContext(domain, subject, new CallbackHandler() {
  public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException {
    for (int i = 0; i < callbacks.length; i++) {
      if (callbacks[i] instanceof NameCallback) {
        ((NameCallback) callbacks[i]).setName(user);
      } else if (callbacks[i] instanceof PasswordCallback && credentials instanceof String) {
       ((PasswordCallback) callbacks[i]).setPassword(((String) credentials).toCharArray());
      } else if (callbacks[i] instanceof CertificateCallback && credentials instanceof X509Certificate) {
        ((CertificateCallback) callbacks[i]).setCertificate((X509Certificate) credentials);
      } else {
        throw new UnsupportedCallbackException(callbacks[i]);
      }
    }
  }
});

For the NameCallback it sets the username that was extracted from the WS UsernameToken profile of the incoming SOAP request. And for the PasswordCallback it sets the corresponding password. If the callback was of type CertificateCallback, it would set the X.509 certificate of the incoming WSSE security header (this is not the case in our example).
Next we jump back into our PropertiesLoginModule.login() method. The two callback objects now contain the client's username and password (as sent with the SOAP request), so we can authenticate them.
The rest is very simple. The username and password extracted from the callback are compared to the user/password combinations loaded from users-passwords.properties and if they don't math an FailedLoginException is raised.

// method PropertiesLoginModule.login() continued
user = ((NameCallback) callbacks[0]).getName();
char[] tmpPassword = ((PasswordCallback) callbacks[1]).getPassword();
if (tmpPassword == null) {
  tmpPassword = new char[0];
}
String password = users.getProperty(user);
if (password == null) {
  throw new FailedLoginException("User does not exist");
}
if (!password.equals(new String(tmpPassword))) {
  throw new FailedLoginException("Password does not match");
}

It is this code where the authentication either succeeds or fails. If authentication fails, a FailedLoginException is raised, causing the call to PropertiesLoginModule.abort(), otherwise PropertiesLoginModule.commit() is invoked. Inside commit() we only add the relevant security roles from groups.properties to the authenticated user and add the entire authenticated principal to the java security Subject:

public boolean commit() throws LoginException {
        principals.add(new UserPrincipal(user));

  for (Enumeration enumeration = groups.keys(); enumeration.hasMoreElements();) {
    String name = (String) enumeration.nextElement();
    String[] userList = ((String) groups.getProperty(name) + "").split(",");
    for (int i = 0; i < userList.length; i++) {
      if (user.equals(userList[i])) {
        principals.add(new GroupPrincipal(name));
        break;
      }
    }
  }
  subject.getPrincipals().addAll(principals);
  ...
}

The Java security Subject now carries all credential information including its user roles.
The list of roles will be required at last to perform authorization.

Note: All these operations are performed from within JbiJAASInterceptor.handleMessage(). This interceptor handles all the complex JAAS authentication part. Thus when this interceptor finishes, the client request is either authenticated or rejected. The remaining CXF interceptors can now be invoked but they won't perform any security related functions.

Finally the
org.apache.servicemix.cxfbc.JbiInvokerInterceptor.handleMessage()
will use the org.apache.servicemix.jbi.security.SecuredBroker to place the constructed JBI message onto the ServiceMix Normalized Message Router (NMR). But before doing so, the SecuredBroker will check if the caller is authorized. So this is the final authorization.

org.apache.servicemix.jbi.security.SecuredBroker.sendExchangePacket()
Before the message is sent to the NMR, the SecuredBroker checks if the caller is authorized. Once the JBI message containing the payload of the SOAP message is put onto the NMR it will be delivered to the target service (in our case the demo's ws-security-cxfse-su component) without performing any security checks. So this authorization is the final security operation in our scenario.

The information about what roles are allowed to invoke on what services are specified in conf/security.xml in your ServiceMix installation. Out of the box there is an <authorizationMap> entry as follows:

<sm:authorizationMap id="authorizationMap">
    <sm:authorizationEntries>
      <sm:authorizationEntry service="*:*" roles="*" />
    </sm:authorizationEntries>
  </sm:authorizationMap>

So out of the box anyone can invoke on any service. This basically turns off authorization. The source code actually checks if roles="*" and bypasses authorization completely:

// inside SecuredBroker.sendExchangePacket()
if (!acls.contains(GroupPrincipal.ANY)) {
   //perform authorization, details omitted.
}

... whereby GroupPrincipal.ANY resolves to "*".
If you want to enable authorization you need to specify one or more authorization entries in security.xml where the role are not just "*", meaning everyone. You would generally list some of the role names from your conf/groups.properties file.
Of course you can fine grain access control to a particular service by specifying the service attribute in the <authorizationEntry> element. The syntax is

 
 <sm:authorizationEntry service="{namespace}:servicename" roles="admin"/>

as typically defined in your xbean.xml service configuration. So in our example, in order to limit access to only the administrator role for the cxf-se service, you would specify:

<sm:authorizationEntry service="{http://apache.org/hello_world_soap_http}:SOAPServiceWSSecurity" 
  roles="admin" />

So this list of authorization entry configuration from security.xml is available to the SecuredBroker. On the other hand it also has access to the javax.security.auth.Subject instance that contains all the clients security credentials including the authenticated user roles and it can extract the service name of the target service to invoke on from the JBI MessageExchange. Hence it is just a question of checking that the user role in the security Subject is allowed to invoke the service named in the MessageExchange. This check will be performed using the authorization configuration from security.xml.
If the caller is not authorized a security exception will be thrown, otherwise the exchanges gets put onto the NMR from where it will be routed to the right service.
That's it. In case the authorization succeeds the JBI message is put onto the NMR from where it will be routed and dispatched to the target service.

Executive Summary:

• The WSS4JInInterceptor runs the WS-Security related functions, extract WS username and password from the SOAP security header and puts it into a WSUsernameTokenPrincipal object.
  
• The JbiJAASInterceptor is the next CXF interceptor to invoke on and checks for the presence of a WSUsernameTokenPrincipal, extracts the credential information and invokes on the JAAS authentication service.


• The JAAS authentication service uses the configured login module class (PropertiesLoginModule as set in conf/login.properties) to delegate the authentication decision.


• The PropertiesLoginModules calls back on the registered JAAS callback handler to retrieve the username/password and authenticates these credentials against the definitions in user-passwords.properties. If authentication succeeds, it assigns the list of roles to this authenticated Subject. The call stack returns all the way back to the JbiJAASInterceptor.


• The JbiInvokerInterceptor dispatches the request and uses the SecuredBroker to send the message exchange to the NMR.


• The SecuredBroker performs authorization based on the user role names stored in the Java security Subject and based on the authorization configuration made in conf/security.xml.

I hope I could spot some light onto the mysteries of authentication/authorization in ServiceMix 3.
The ServiceMix documentation does unfortunately not document these concepts in much detail.

Looking forward to any feedback from you.

Slower performance with the new servicemix-jms endpoints?

There are two servicemix-jms components available in the later versions of ServiceMix 3. The traditional jms endpoint that is configured using <jms:endpoint> and the new jms endpoint that uses <jms:consumer>; and <jms:provider>. The new JMS endpoints reuse much of Spring JMS. Very recently I noticed a pretty large performance difference between these two JMS consumers, in the way that the new servicemix-jms consumer was much slower than the traditional consumer while using a pretty default configuration for both.
The reason for that is that out of the box the new JMS endpoint is configured to use the Spring JMS DefaultMessageListenerContainer without any caching, so all JMS resources get re-created and closed for every message being sent or received. This of course degrades performance a lot.

There are two simple solutions:

1) Configure caching for this DMLC

  <jms:consumer service="test: NewJMSConsumer" 
    endpoint="endpoint" 
    targetservice="test:bean" 
    targetendpoint="endpoint" 
    destinationname="queue/TEST.FOO" 
    connectionfactory="#AMQConnectionFactory" 
    cachelevel="3">

Using cacheLevel="3" (CACHE_CONSUMER) is the highest cache option and caches the pretty much all JMS resources for each listener thread (JMS connection, session and message consumer). Other levels are 2 (CACHE_SESSION ), 1 (CACHE_CONNECTION) and 0 (CACHE_NONE). If no cacheLevel is specified, CACHE_NONE is used!

2) Or configure for a different Spring JMS listener

  <jms:consumer service="test:NewJMSConsumer" 
    endpoint="endpoint" 
    targetservice="test:bean" 
    targetendpoint="endpoint" 
    destinationname="queue/TEST.FOO" 
    connectionfactory="#AMQConnectionFactory" 
    listenertype="server">

The ServerSessionMessageListenerContainer type internally uses a ServerSessionPool (which as the name suggests pools the server sessions) and also results in reasonable performance. However this listener type might not be suitable for all deployments.

Both options should boost performance and perform equally fast as the old JMS endpoint. These jms consumer attributes are both documented but it is easy to overlook them and just use the default configuration.
I recommend solution 1, using the DefaultMessageListenerContainer and enable caching.

Finally there seems no reason why caching could not be turned on by default. So I raised JIRA SM-1841 at Apache.