ActiveMQ Network Bridge to Master/Slave broker pair

This article got updated on 10/20/11 in order to include the latest changes to the failover protocol for version 5.6.

There are scenarios where in a network of brokers each broker instance needs to be highly available. For example in hub and spoke architectures it might not be tolerable to be disconnected from the hub broker for a certain amount of time. This would prevent processing messages from spokes in real-time by consumers attached to the hub broker.

Consider an ActiveMQ network of brokers configuration of two nodes where each node also needs to be highly available. To achieve High Availability, each node needs to have a slave broker attached like in the following picture:

As shown in the above figure, Master Broker 1 needs to be able to create a network bridge to either Master Broker 2 or in case it is unavailable, Slave Broker 2.
Likewise if Master Broker 1 dies, Slave Broker 1 will take over and it also needs to be able to connect to either Master or Slave of Broker 2.
That way you can achieve high availability of all broker instances in a network of brokers. You already guess that such configuration can become really complex for a larger number of brokers in the network (also depending on the cluster topology).


The configuration of Master/Slave is decently well explained in the ActiveMQ documentation. But how do we configure the brokers network connector so that the network bridge is able to failover to a slave if the master crashes?

When trying to setup a bridge from Master Broker 1 to the master/slave pair of Broker 2, a configuration like

<networkConnector name="NC_1"
    uri="static://(tcp://broker2Master:61616,tcp://broker2Slave:61616)"/>

won’t be ideal. Because of the static: url list, the network connector will try to connect to both Master Broker 2 and Slave Broker 2. It will connect to Master Broker 2 just fine but will also retry connecting to Slave Broker 2 forever. Any failed attempt will be logged inside Broker 1 with a warning

WARN | Could not start network bridge between vm://localhost and tcp://broker2Slave:61616 due to: java.net.ConnectExcption: Connection refused.

This does not only pollute the log file and makes it harder to spot any other issues in the log but it also takes some amount of CPU every couple of seconds for trying a reconnect.

Side note: When using static:// in the network connector uri, lost connections will be re-established automatically. On this regards it works like the failover protocol but it tries to connect to all urls in the list.

Instead of trying to connect to all brokers specified in the above network connector uri list (i.e. Master Broker 2 and Slave Broker 2), we only want the bridge to connect to one broker of the list. Therefore a better idea is to use the failover:// protocol inside the network connector configuration. We only want the Master Broker 1 to connect to either Master Broker 2 or Slave Broker 2 but not both. So the configuration now becomes

<networkConnector name="NC_1"
   uri="static:failover:(tcp://broker2Master:61616,tcp://broker2Slave:61616)?randomize=false"/>

By also setting randomize=false we will always try connecting to Master Broker 2 first, before attempting to connect to the Slave Broker 2. This configuration will keep trying to connect to one of the two brokers until a connection gets established. It will not try to connect to both brokers in the list!

This configuration however still has a problem.
In case of a connection loss the failover transport will keep trying to reconnect to the specified urls transparently and will not propagate any exceptions up to the higher layer. So if after a crash of the Master 2 broker the Slave 2 broker has finally started up, the tcp transport connection will get re-established transparently by the failover transport in Master Broker 1 without flagging the loss of connection to any higher layers. That implies the broker’s discovery agent that is responsible for creating the network bridge will never be notified and hence will never try to re-create the network bridge (which involves exchanging a couple of messages between brokers). 
You can confirm this in jconsole, looking at the bridge MBean. The name of the MBean will refer to the previous master broker whereas the RemoteAddress property got updated with the new master broker.
Prior to version 5.6 the bridge would most likely not work anymore. Things have improved in 5.6 thanks to the fix of AMQ-3542 and the bridge may continue to work but the original problem remains. The bridge need to get re-established after failing over.

So what is needed to re-establish the network bridge after a failover? The solution is to use the failover transport for trying to connect to the urls in the connection list without any reconnects. 
When the tcp connection of the bridge is lost, the failover protocol will not try to reconnect but raise an exception to the DiscoveryAgent. The agent will clean up its bridge and ask the underlying failover transport to re-establish a connection to either master or slave of Broker 2. Now the failover transport will try all urls in its list only once, it won't attempt any reconnects on its own. If it still can't connect, it raises an exception back to the agent again, which after a timeout will ask the failover transport again to reconnect. This continues until the failover transport succeeds in connecting to one of the specified urls. Once the tcp connection is established, the DiscoveryAgent can recreate the bridge based on the new connection.
By not letting the failover transport reconnect on its own, the DiscoveryAgent is now aware that the connection of the network bridge got lost (as the error was propagated) and will re-establish the network bridge once the connection got restored. 

The failover transport needs to be explicitly configured to not reconnect on its own but raise an exception instead. By default it will reconnect forever.
From version 5.6 onwards, you should use the failover property maxReconnectAttempts=0 for that reason.

So the network bridge configuration finally becomes (for versions 5.6 onwards):

<networkConnector name="NC_1" uri="static:failover:(tcp://broker2Master:61616,tcp://broker2Slave:61616)?randomize=false&maxReconnectAttempts=0"/>


Also from version 5.6 onwards you will be able to use "masterslave:" instead of "static:failover:()". So the above example becomes

<networkConnector name="NC_1" uri="masterslave:(tcp://broker2Master:61616,tcp://broker2Slave:61616)"/>

See AMQ-3564 for more details but "masterslave:" simply maps to 
"static:failover:()?randomize=false&maxReconnectAttempts=0"

Prior to version 5.6 you cannot really configure for no reconnects. A value of 0 means "reconnect forever". From version 5.6 the value 0 means "do not reconnect". See the updated failover transport reference.


So for versions < 5.6 instead use maxReconnectAttempts=1 and allow one reconnect, which by default happen within 10 milliseconds (unless configured otherwise using initialReconnectDelay property of the failover transport). This is generally to short for the slave broker to take over. 



The network bridge configuration for Slave Broker 1 would be the same as it also needs to connect to either Master Broker 2 or Slave Broker 2. 

Summary: When trying to setup a network bridge to a master/slave broker pair, use the failover transport with maxReconnectAttempts=0.

Side note: When the remote broker is shutdown gracefully, the network bridge will get unregistered and closed down so that the discovery agent will always be aware of it. In that case the network bridge will be re-established correctly when the remote broker gets restarted even when not setting maxReconnectAttempts. But it won’t help in case the remote broker crashes.

Error handling in Camel for JMS consumer endpoint

As of version 2.0 Camel now uses the DefaultErrorHandler out of the box. It offers a more Java-like error handling in the way that any exceptions that occur while routing the message will be propagated back to the caller while also ending the Exchange immediately. See the Camel error handling documentation for more details.

This is great for most use cases but not the best choice for every scenario. For illustration purposes lets consider the example of using Camel as a JMS bridge (although the following applies to any Camel route starting with a jms consumer endpoint).

Here’s a simple Camel route definition that routes messages from Apache ActiveMQ to IBM WebSphere MQ:

<camelContext id="camel" xmlns="http://camel.apache.org/schema/spring">
  <route>
    <from uri="activemq:queue:GatewayToWebSphereMQ"/>
    <to uri="webspheremq:queue:FromActiveMQ"/>
  </route>
</camelContext>

I leave out the Spring bean definition of the activemq and webspheremq JMS components for simplicity.
This route will forward any messages sent to the ActiveMQ queue "GatewayToWebSphereMQ" to a queue called "FromActiveMQ" in WebSphere MQ.

Now lets see what happens if WebSphere MQ is down for whatever reasons.
If a new message is put on the GatewayToWebSphereMQ, it will be picked up by the camel-activemq component. This component uses AUTO_ACKNOWLEDGE mode by default. That means Camel will receive the message from the broker and ack it straight away, before routing the message any further. From the point of view of the broker, the message has been consumed and in fact the message is now entirely in the hands of Camel.

So after acking the message, Camel will now try to route the message, i.e. sending it to WebSphere MQ. But the other end is down, so we won’t get a tcp connection established. Instead some sort of a socket exception will be thrown.
This is where the error handling in Camel comes into play now.

From Camel 2.0 onwards the DefaultErrorHandler will be called. The default behavior of this error handler is to propagate the error back to the caller. In this example this isn't possible as the camel-jms component has already consumed and acked the message. So it cannot raise the exception to the broker nor can we put the message back on top of the queue. Instead what happens is the message gets discarded (without being stored anywhere) after logging the error. So the message is basically lost. This is certainly not ideal if you cannot afford to loose messages.

There are a few solutions:

1) The DefaultErrorHandler will by default not try to redeliver the message. You can configure the error handler for a different redelivery policy so that it attempts to redeliver the message a couple of times before giving up in the hope that WebSphere will have restarted within that time frame. However when finally giving up on the retries, the message would still be discarded and lost.


2) If you can't afford to loose messages the better solution is to use a different Camel error handler, i.e. the Dead Letter Channel. This handler will move the message to a configurable dead letter queue if it cannot be routed.
Here is a sample configuration for a dead letter channel:

<bean id="myDeadLetterErrorHandler" class="org.apache.camel.builder.DeadLetterChannelBuilder">
  <property name="deadLetterUri" value="activemq:queue:ActiveMQ.DLQ"/>
  <property name="redeliveryPolicy" ref="myRedeliveryPolicyConfig"/>
</bean>

<bean id="myRedeliveryPolicyConfig" class="org.apache.camel.processor.RedeliveryPolicy">
  <property name="maximumRedeliveries" value="3"/>
  <property name="redeliveryDelay" value="5000"/>
</bean>

Now lets tell Camel to use this error handler instead of the default handler:

<camelContext id="camel" xmlns="http://camel.apache.org/schema/spring">
  <route errorHandlerRef="myDeadLetterErrorHandler">
    <from uri="activemq:queue:GatewayToWebSphereMQ"/>
    <to uri="webspheremq:queue:inbound/inbox"/>
  </route>
</camelContext>

This example configures the dead letter channel with a custom redelivery policy. Camel will now retry every message three times with a 5 seconds delay. If delivery is still unsuccessful thereafter, the msg gets moved to the queue "ActiveMQ.DLQ" in ActiveMQ. If the connectivity problem to WebSphere MQ is only short term, a properly configured redelivery policy may prevent moving any messages to a dead letter queue.
You can configure for any other dead letter queue and in fact it does not necessarily have to be a JMS queue, as "seda:errorqueue" will also work.

This configuration of a Camel error handler will never loose any persistent messages! However you will need to think of a strategy what to do with messages ending up on a dead letter queue (e.g. manually re-route them back to the original queue after the connection to WebSphere MQ got restored).


3) A third possible solution would be to use a transacted Camel route. For transacted routes theTransactionErrorHandler is used. The camel-jms endpoint is a transaction capable endpoint, and so do ActiveMQ and WebSphere MQ support transactions. The entire Camel route above could therefore spawn a single transaction. If there are any errors encountered within the transaction (e.g. while trying to send the message to WebSphere MQ), the transaction will be rolled back and the message is moved back to the original queue again (in fact, it never leaves the queue).
Now you require an appropriate redelivery policy configuration inside ActiveMQ (e.g. try to redeliver the message up to 5 times before moving the message to a dead letter queue "ActiveMQ.DLQ").
Further as the Camel route involves two different JMS endpoints, you would need to configure Camel for XA transactions, involving an XA transaction monitor. XA transactions also have an impact on performance and might not always be needed. The Camel Transaction Guide on FuseSource.com has some really good chapters on configuring Camel for XA transactions.

The second solution outlined above might be the easiest solution that guarantees messages won’t get lost.

java.io.InvalidClassException: NoSuchCustomerException; local class incompatible: stream classdesc serialVersionUID = 20110530114741

While working on a camel demo that routes a soap message in POJO mode from a cxf-consumer endpoint via jms to a camel processor, I ran into the above exception when the camel processor returned a soap fault. The full error reads [1].

The Camel route definition is as simple as

<route id="CXF-to-Queue">
  <from uri="cxf:bean:customer-ws?dataFormat=POJO"/>
  <inOut uri="activemq:queue:lookupCustomer?jmsMessageType=Object&transferException=true"/>
</route>

<route id="Queue-to-Processor">
  <from uri="activemq:queue:lookupCustomer?jmsMessageType=Object&transferException=true" />
  <process ref="lookupCustomer"/>
</route>

I ran each Camel route on a different machine with both routes connecting to the same broker. Further all msgs were passed as ObjectMessages in JMS (but you would get the same issue with JMS ByteMessages or any binary serialization of the exception).
The SOAP fault (raised by the lookupCustomer processor) was marshaled into the JMS ObjectMessage correctly and sent back on the reply-to queue, but the receiving camel-jms endpoint (my first route) had problems unmarshaling the data and raised the InvalidClassException as shown above.

After some digging I got to learn that since I re-compiled my demo on the other machine, a new and different serialVersionUID was generated for NoSuchCustomerException.java:

@WebFault(name = "NoSuchCustomer", 
  targetNamespace = "http://demo.fusesource.com/wsdl/CustomerService/")
public class NoSuchCustomerException extends Exception {
    public static final long serialVersionUID = 20110530174707L;
    ... 
}

This serial version uid is generated based on a time stamp by default! So each time wsdl2java is re-run, a different uid will be generated. This is often the case with mvn based projects. As I compiled my demo on both machines (at different times), the uid differed.

There was an easy fix to it however, which is to tell the wsdl2java compiler to generate the serial version uid based on the fully qualified classname (-useFQCNForFaultSerialVersionUID). This will always generate the same uid for the same fault definition no matter how often I run wsdl2java. See http://cxf.apache.org/docs/wsdl-to-java.html for more information.

IMHO, -useFQCNForFaultSerialVersionUID should be the default in order to avoid such problems.


You won't have this problem when using SOAP/HTTP as the transport. The JAXB marshaling won't marshal the uid.
This problem only arises when using binary transports such as JMS.

[1] full error
org.apache.camel.RuntimeCamelException: Failed to extract body due to: javax.jms.JMSException: Failed to build body from bytes. Reason: java.io.InvalidClassException: com.fusesource.demo.wsdl.customerservice.NoSuchCustomerException; local class incompatible: stream classdesc serialVersionUID = 20110530114741, local class serialVersionUID = 20110530112224. Message: ActiveMQObjectMessage {commandId = 61, responseRequired = false, messageId = ID:nbwfhtmielke-1657-1306750363546-3:1:1:1:4, originalDestination = null, originalTransactionId = null, producerId = ID:nbwfhtmielke-1657-1306750363546-14:1:1:1, destination = queue://reply.test1, transactionId = null, expiration = 0, timestamp = 1306755695062, arrival = 0, brokerInTime = 1306755696022, brokerOutTime = 1306755696678, correlationId = ID-XPS-53828-1306755653494-0-6, replyTo = queue://reply.test1, persistent = true, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = org.apache.activemq.util.ByteSequence@ed92dbb, marshalledProperties = org.apache.activemq.util.ByteSequence@5449579a, dataStructure = null, redeliveryCounter = 0, size = 0, properties = {Content_HYPHEN_Type=text/xml;charset=UTF-8, operationNamespace=http://demo.fusesource.com/wsdl/CustomerService/, operationName=lookupCustomer, Host=localhost:10443, SOAPAction="http://www.example.org/CustomerService/lookupCustomer", User_HYPHEN_Agent=Jakarta Commons-HttpClient/3.1, CamelJmsDeliveryMode=2, accept_HYPHEN_encoding=gzip,deflate}, readOnlyProperties = true, readOnlyBody = true, droppable = false}
at org.apache.camel.component.jms.JmsBinding.extractBodyFromJms(JmsBinding.java:158)[camel-jms-2.5.0-fuse-00-00.jar:2.5.0-fuse-00-00]
at org.apache.camel.component.jms.JmsMessage.createBody(JmsMessage.java:183)[camel-jms-2.5.0-fuse-00-00.jar:2.5.0-fuse-00-00]
at org.apache.camel.impl.MessageSupport.getBody(MessageSupport.java:41)[camel-core-2.5.0-fuse-00-00.jar:2.5.0-fuse-00-00]
at org.apache.camel.component.jms.reply.ReplyManagerSupport.processReply(ReplyManagerSupport.java:112)[camel-jms-2.5.0-fuse-00-00.jar:2.5.0-fuse-00-00]
at org.apache.camel.component.jms.reply.TemporaryQueueReplyHandler.onReply(TemporaryQueueReplyHandler.java:52)[camel-jms-2.5.0-fuse-00-00.jar:2.5.0-fuse-00-00]
at org.apache.camel.component.jms.reply.PersistentQueueReplyHandler.onReply(PersistentQueueReplyHandler.java:45)[camel-jms-2.5.0-fuse-00-00.jar:2.5.0-fuse-00-00]
at org.apache.camel.component.jms.reply.PersistentQueueReplyManager.handleReplyMessage(PersistentQueueReplyManager.java:84)[camel-jms-2.5.0-fuse-00-00.jar:2.5.0-fuse-00-00]
at org.apache.camel.component.jms.reply.ReplyManagerSupport.onMessage(ReplyManagerSupport.java:98)[camel-jms-2.5.0-fuse-00-00.jar:2.5.0-fuse-00-00]
at org.springframework.jms.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:560)[spring-jms-3.0.5.RELEASE.jar:3.0.5.RELEASE]
at org.springframework.jms.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:498)[spring-jms-3.0.5.RELEASE.jar:3.0.5.RELEASE]
at org.springframework.jms.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:467)[spring-jms-3.0.5.RELEASE.jar:3.0.5.RELEASE]
at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.doReceiveAndExecute(AbstractPollingMessageListenerContainer.java:325)[spring-jms-3.0.5.RELEASE.jar:3.0.5.RELEASE]
at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.receiveAndExecute(AbstractPollingMessageListenerContainer.java:263)[spring-jms-3.0.5.RELEASE.jar:3.0.5.RELEASE]
at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.invokeListener(DefaultMessageListenerContainer.java:1058)[spring-jms-3.0.5.RELEASE.jar:3.0.5.RELEASE]
at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.executeOngoingLoop(DefaultMessageListenerContainer.java:1050)[spring-jms-3.0.5.RELEASE.jar:3.0.5.RELEASE]
at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.run(DefaultMessageListenerContainer.java:947)[spring-jms-3.0.5.RELEASE.jar:3.0.5.RELEASE]
at java.lang.Thread.run(Thread.java:636)[:1.6.0_20]

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().