ServiceMix with LDAP based authentication

It’s possible to configure LDAP based authentication in ServiceMix. We recently added a complete tutorial in the FUSE ESB Security Guide. It provides step-by-step instructions and screenshots based on the Open Source LDAP server ApacheDS.
Anyone having to configure LDAP in SMX, check out chapter 4 "LDAP Authentication Tutorial".

How to SSL enable the ServiceMix web console?

ServiceMix 4 comes with a useful web console. It needs to be installed manually; it is not deployed out of the box:

karaf@root> features:install webconsole

... and thereafter the console can be accessed using URL:
http://localhost:8181/system/console/bundles
(applies to 4.3.1, URL might differ on other versions of SMX).

In order to secure the console to use HTTPS, it is necessary to create a file

$KARAF_HOME/etc/org.ops4j.pax.web.cfg

and configure it using any of the property keywords defined in this WebContainerConstants class.

Here is a possible example:

# configures the SMX Web Console to use SSL
#
# @SeeAlso: https://github.com/ops4j/org.ops4j.pax.web/blob/master/pax-web-api/src/main/java/org/ops4j/pax/web/service/WebContainerConstants.java
# for possible configuration properties

org.osgi.service.http.enabled=false
org.osgi.service.http.port=8181

org.osgi.service.http.secure.enabled=true
org.osgi.service.http.port.secure=8183

org.ops4j.pax.web.ssl.keystore=/path/to/keystore.ks
org.ops4j.pax.web.ssl.keystore.type=JKS
org.ops4j.pax.web.ssl.password=blah
org.ops4j.pax.web.ssl.keypassword=bluh
org.ops4j.pax.web.ssl.clientauthwanted=false
org.ops4j.pax.web.ssl.clientauthneeded=false

Make sure to access the web console using https:// after applying this configuration.
;-)

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.