Ramblings on pretty much anything technical.

Securing the ActiveMQ web console using LDAP based authentication

2011-12-22T09:08:00.000+01:00

In my previous blog post I described how to configure LDAP based authentication and authorization in ActiveMQ while also allowing anonymous access to certain destinations in the broker. That previous post as well as this post is based on the LDAP tutorial in the ActiveMQ Security Guide provided by Fusesource.com.

In this article I want to expand on the previous post and show how to secure the ActiveMQ web console so that web users will be authenticated against user information stored in LDAP. All the configuration from the previous post can extended in order to secure the web console but I will provide full configuration here as well so that there is no need to copy and paste config snippets from the previous article.
I recommend the LDAP tutorial as a prerequisite to this article as it populates the LDAP server with the right security information (users, passwords, groups and roles), which is needed if you want to follow the steps of this blog post.

The ActiveMQ web console internally uses Jetty as the web container. So securing the web console basically means configuring the Jetty HTTP server to authenticate any web clients. The way how Jetty gets secured is completely independent from the way how ActiveMQ gets secured. However both support JAAS so the actual configuration principles are the same. Jetty also supports LDAP based authentication through its own JAAS LoginModule class org.eclipse.jetty.plus.jaas.spi.LdapLoginModule.

Prerequisites before taking off

The LDAP tutorial in the ActiveMQ Security Guide (version 5.5) makes some configuration that will not work quite well with the JAAS LDAP login module used by the ActiveMQ web console (i.e. Jetty). Two configuration changes are required in your LDAP user and group data to successfully authenticate any web users via Jetty's LdapLoginModule. Here are the required changes to your LDAP data:

1) Store passwords in plain text in LDAP:
The LDAP tutorial instructs you to store any user passwords as SHA hashes. See steps 18 and 19 of the Add User Entries part of the tutorial (and Figure 6.7). The LdapLoginModule from Jetty reads that password from LDAP as it is (i.e. SHA hash) and tries to match it against the password supplied by the web user, which is in plain text. So the SHA hashed password stored in LDAP is string compared against the plain text password, which will obviously fail.
Rather than storing the password as SHA hash in LDAP, store it in plain text.

This needs to be done for all users.
Storing the password in plain text will not change the ability of the ActiveMQ LDAPLoginModule class to authenticate users. This LoginModule takes a different approach. Rather than retrieving the password and comparing it against the password given by the user, it tries to bind the user (an operation on the LDAP server) whereby the password gets verified by the LDAP server and not by the LoginModule class.
Perhaps there is a way to configure the Jetty LdapLoginModule to also work with SHA hashed password, however I did not find it (also tried DIGEST based authentication but that did not help).

2) User to group mapping in LDAP must use users full dn + basedn name:

Steps 30-31 and figure 6.8 of the Add User Entries part of the LDAP tutorial explain how to make users member of a particular group. It uses the member attribute of the groupOfNames LDAP class and the mapping to a user is done via the users uid attribute, like in this example mapping that is taken from the LDAP tutorial:
member=uid=jdoe

This won't work with the Jetty LdapLoginModule implementation, as it queries for a users group using the user's full name (i.e. uid + base dn), e.g.

member=uid=jdoe,ou=User,ou=ActiveMQ,ou=system

and as a result won't find any user groups if the group members are only specified using the uid (i.e. jdoe).
To overcome this difference in the LDAP login module implementations, it is necessary to always provide the dn + basedn of the user that you want to add to a group.
E.g. rather than setting the members of a group like this ((see Figure 6.8 of the LDAP tutorial)

use this

This change of users to groups wiring however requires an update on the ActiveMQ login module configuration in login.config from
roleSearchMatching="(member=uid={1})"

to
roleSearchMatching="(member=uid={1},ou=User,ou=ActiveMQ,ou=system)"

That way both, the LDAP LoginModule of ActiveMQ and Jetty will be able to retrieve the groups (roles) a user is in.
With these two changes to the LDAP data we are ready to go and configure the web console for LDAP based authentication.

Steps to securing the web console

All of the Jetty configuration in ActiveMQ is stored in $ACTIVEMQ_HOME/conf/jetty.xml. Its Spring config already instantiates a security handler but authentication is turned off by default via the authenticate=false property. It needs to be enable in this bean:

bean>

Changing the "authenticate" property to "true" is required but isn't enough. Any authenticated user needs to have the role specified by the "roles" attribute in order to get authorized. The administrator role name in LDAP is called "admins" but the Jetty securityConstraint bean uses "admin" out of the box. It is necessary to change the value of the "roles" property to "admins" as in the above example. A fully configured jetty.xml is provided at the end of this article.

Out of the box Jetty is configured to use a HashLoginService class, which simply reads the user credentials from a text file ${activemq.base}/conf/jetty-realm.properties. In order to authenticate against an LDAP server we need to configure a JAAS LoginService class:

bean>

This JAASLoginService is configured for a LoginModuleName called "jetty-ldap". This name refers to the JAAS login module configuration in login.config. In the previous article I already provided a login.config to be used by ActiveMQ. You can simply add the following configuration to that same login.config:

/**

This LoginModule configuration is only used by Jetty when

authentication of the web console is enabled.

jetty-ldap {

org.eclipse.jetty.plus.jaas.spi.LdapLoginModule required

debug="true"

contextFactory="com.sun.jndi.ldap.LdapCtxFactory"

hostname="localhost"

port="10389"

bindDn="uid=admin,ou=system"

bindPassword="secret"

authenticationMethod="simple"

forceBindingLogin="false"

userBaseDn="ou=User,ou=ActiveMQ,ou=system"

userRdnAttribute="uid"

userIdAttribute="uid"

userPasswordAttribute="userPassword"

userObjectClass="inetOrgPerson"

roleBaseDn="ou=Group,ou=ActiveMQ,ou=system"

roleNameAttribute="cn"

roleMemberAttribute="member"

roleObjectClass="groupOfNames"

authenticated="true";

};

Finally Jetty needs to be told where to find this login.config file via a Java property to the JVM.
The easiest way is to add this propery to the bin/activemq script, e.g.

ACTIVEMQ_OPTS="$ACTIVEMQ_OPTS_MEMORY -Dorg.apache.activemq.UseDedicatedTaskRunner=true -Djava.util.logging.config.file=logging.properties -Djava.security.auth.login.config=login.config"

That's about it. Assuming the data in LDAP is setup according to the prerequisites listed above, the ActiveMQ web console should now be secured. Trying to access http://localhost:8161 should now request a username and password from you which is then authenticated against the LDAP server.

Note we did not need to change the broker configuration (with the exception of pre-requisite #2) as the authentication done in the web console (based on Jetty) is completely independent of the authentication done by ActiveMQ.

Links to all files used in this post:
jetty.xml
login.config
activemq-ldap-with-anon-access.xml

The relevant documentation that helped me on this subject:
Jetty JAAS Tutorial
Jetty Realms Tutorial
Jetty 7 JavaDoc

Let me know if this post was helpful.

ActiveMQ: LDAP based authentication and authorization

2011-12-05T17:24:00.001+01:00

The FuseSource ActiveMQ Security Guide has two great chapters on how to configure ActiveMQ for authentication and authorization against an LDAP server. Chapter 5 even has a tutorial that contains step-by-step instructions on how to configure your LDAP server and ActiveMQ based on ApacheDS, an open-source LDAP server. If you need to secure your ActiveMQ broker, then I highly recommend this documentation.

The tutorial in the Security Guide shows the relevant configuration needed so that every JMS connection into the broker is authenticated and checked for authorization against the security information stored in an LDAP server.

Sometimes you may want to allow anonymous access to certain destinations on your broker that are not critical while securing access to your critical destinations. In this post I like to outline a possible solution for such use-case. I will assume the reader is generally familiar with LDAP based authentication and authorization in ActiveMQ. If not, I suggest to first consult the ActiveMQ Security Guide.

When configuring ActiveMQ for JAAS based authentication against an LDAP server, the file login.config could read as follows (pretty much a copy-and-paste from the FuseSource Security Guide):

/** JAAS LoginModule that uses LDAP basedauthentication.

Every connection into broker must supplyusername and password

for authentication to succeed. Anonymousaccess not allowed.

LDAPLogin {

org.apache.activemq.jaas.LDAPLoginModulerequired

debug=true

initialContextFactory=com.sun.jndi.ldap.LdapCtxFactory

connectionURL="ldap://localhost:10389"

connectionUsername="uid=admin,ou=system"

connectionPassword=secret

connectionProtocol=""

authentication=simple

userBase="ou=User,ou=ActiveMQ,ou=system"

userSearchMatching="(uid={0})"

userSearchSubtree=false

roleBase="ou=Group,ou=ActiveMQ,ou=system"

roleName=cn

roleSearchMatching="(member=uid={1})"

roleSearchSubtree=false;

};

The broker then needs to use the JAAS authentication plug-in referencing the LDAPLogin authentication realm:

</plugins>

Using such configuration, every JMS connection will be authenticated using the LDAPLoginModule, no matter if it provides a username and password. Needless to say the authentication will fail if no username/password is supplied. Anonymous access won't be allowed in this configuration.

So in order to also allow anonymous access to the broker we can leverage the ActiveMQ GuestLoginModule.
The GuestLoginModule allows JMS connections that are not configured for username/password to still access a secured broker. It is typically used in conjunction with other JAAS login modules and basically successfully authenticates a JMS connection using a configurable username and group name. Here is a sample configuration of this login module:

org.apache.activemq.jaas.GuestLoginModulesufficient

debug=true

org.apache.activemq.jaas.guest.user="guest"

org.apache.activemq.jaas.guest.group="guests";

Authentication will succeed even if no username/password was supplied by the JMS client. Clients will be authenticated as user "guest" and belong to the group "guests". As you see, these names are configurable.

In JAAS configuration, multiple login modules can be combined in one JAAS authentication realm. They will then be tried in order.

The basic idea is to configure authentication so that the broker first invokes the GuestLoginModule, before trying the LDAPLoginModule. Here is the configuration:

/** JAAS LoginModule configuration thatcombines LDAPLoginModule with

GuestLoginModule.

LoginModules can be combined in JAAS.

Seehttp://fusesource.com/docs/broker/5.5/security/Auth-JAAS-GuestLoginModule.html

for more information.

LDAPLogin-with-Anon-Access {

org.apache.activemq.jaas.GuestLoginModule sufficient

debug=true

credentialsInvalidate=true

org.apache.activemq.jaas.guest.user="guest"

org.apache.activemq.jaas.guest.group="guests";

org.apache.activemq.jaas.LDAPLoginModulerequisite

debug=true

initialContextFactory=com.sun.jndi.ldap.LdapCtxFactory

connectionURL="ldap://localhost:10389"

connectionUsername="uid=admin,ou=system"

connectionPassword=secret

connectionProtocol=""

authentication=simple

userBase="ou=User,ou=ActiveMQ,ou=system"

userSearchMatching="(uid={0})"

userSearchSubtree=false

roleBase="ou=Group,ou=ActiveMQ,ou=system"

roleName=cn

roleSearchMatching="(member=uid={1})"

roleSearchSubtree=false;

};

</plugins>

The GuestLoginModule config above uses an additional property credentialsInvalidate that when set to "true" will only authenticate requests that do not have a username/password supplied. Each connection without username gets authenticated as user "guest" and belongs
to the group "guest".
Any other JMS connections that contain username and password will be authenticated using the next login module; the LDAP LoginModule. So its important to set credentialsInvalidate=true as otherwise the GuestLoginModule authenticates all requests no matter whether or what username/password is supplied.

Authenticating anonymous connections is only half the story. Without further authorization anonymous users would be allowed all operations on the broker, which is definitely not what we want.
With additional authorization anonymous access to the broker can be restricted to only specific destinations with restricted rights on each destination.
The FuseSource Security Guide also explains how to configure the LDAP Authorization plug-in and how to create groups and destinations in the LDAP server. A step-by-step guide for adding all required entries is given in chapter 5.
Based on the configuration proposed in that documentation it is additionally also necessary to

define all those destinations in LDAP that anonymous access will be allowed on and
grant the required permissions to the group "guests" to these destinations.

If for example anonymous access should be allowed to queue "example.A", then the destination "example.A" needs to be defined in LDAP. Further "read", "write" and potentially "admin" privileges need to be given to the group "guests" on this destination.
This procedure needs to be applied to all destinations that should allow anonymous access.

Finally, every new connection into the broker triggers some advisory messages. Without giving "guests" access to these advisory topics, the connection will fail, typically with an error similar to this:

java.lang.SecurityException:User null is not authorized to create: topic://ActiveMQ.Advisory.Connection

It is necessary to also grant "admin", "write" and potentially "read" rights for the group "guests" to ActiveMQ.Advisory topics.

This will allow for anonymous access on certain destinations while enforcing username/password based authentication on all other destinations in the broker.

Links to all files used in this post:
login.config
activemq-ldap-with-anon-access.xml
activemq-ldap.xml - no anonymous access

Consuming Topic messages in Spring and Camel

2011-11-14T16:25:00.001+01:00

When using a Spring to consume topic messages, make sure tocache the JMS consumer somewhere. Otherwise your consumer may not receive all messages.

Consider the following Camel route definition that consumesmessages from a JMS topic.

<beanid="JmsConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory" >

</bean>

</bean>

<beanid="jms" class="org.apache.camel.component.jms.JmsComponent" >

</bean>

...

</route>

</camelContext>

The camel-jms component is configured to use a SpringsSingleConnectionFactory and a cache level of CACHE_SESSION. This combinationwill cause problems!

When using Springs SingleConnectionFactory, it will not cachethe consumer. This ConnectionFactory only reuses the same connection but doesnot cache any other JMS resources on top of the connection (i.e. session andconsumer).

At the same time Springs DefaultMessageListenerContainer(which is used by Camel to consume messages from a JMS broker) does not cachethe consumer either as its cache level is set to CACHE_SESSION. The result ofthis configuration is that a new JMS consumer instance get created by SpringsDMLC prior to requesting the next message from the JMS broker. After themessage has been dispatched and processed, the consumer is destroyed (andunregistered from the broker).
In addition a JMS topic works differently from a JMS queue in the way that ifthere is no subscriber registered, the broker will discard the topic message.As with the above configuration the JMS consumer gets recreated for everymessage by Spring (and registered in the JMS broker), there are certain time windowswhen the broker does not have a topic subscriber registered and so it discardsthe messages it receives. The result will be that your Spring JMS topicconsumer will not receive all of the messages from the broker.

When consuming topic messages in Spring, it is necessary to cachethe JMS consumer. Either by using a connection factory that supports consumercaching like the Spring CachingConnectionFactory or by configuring the cachelevel of the DMLC to use CACHE_CONSUMER.

Please note that the ActiveMQ PooledConnectionFactory doesnot cache consumers!

This only applies to topic consumers in Spring. Topicproducers will not use these cache level settings and hence don't have thisproblem.

Loosing messages despite of sending within a transaction

2011-11-07T19:24:00.000+01:00

Last week I made an interesting observation. I was able toloose persistent queue messages that were sent in a transaction to an ActiveMQbroker.

I believe its best to share the findings.

SETTING THE SCENE

Suppose you have a MessageProducer that sends messages to anActiveMQ broker inside a transaction. This producer could be anything, a plainJava JMS Producer client, a Spring JMS producer or as in my case a Camel route.It also does not matter if the producer sends one or more messages within thesame transaction.

Next the ActiveMQ broker instance is configured for aspecific limit. In my test that limit was fairly low inorder to reproduce the problem quickly. In addition I did set the propertysendFailIfNoSpace so that the broker raises an exception back to the produceronce any of its global limits have been reached. So the configuration I usedreads

</memoryUsage>

</storeUsage>

</tempUsage>

</systemUsage>

Without sendFailIfNoSpace="true" ActiveMQ wouldblock the producer until additional space is available on the broker. Thiswould most likely not reproduce the problem I am talking about.

I ran the ActiveMQ broker with these settings, kicked off myCamel route and let it do what it can do best: route messages (which arefinally sent to a queue on my ActiveMQ broker). Btw, I did not have anyconsumers attached to the broker; only the Camel route producer was connected. Stillthe same problem can occur with slow consumers.

And finally my producer did not register an exceptionlistener. In my tests the producer was a Camel route and Camel currently doesnot register an exception listener.

THE PROBLEM

Now, the problem starts when the broker reaches its configuredlimits. Because of the property sendFailIfNoSpace="true", it will notblock the producer indefinitely but throw an exception back to the producer. Inmy test I was hitting the storeUsage limit of 64 MB.

On the other hand when sending messages in a transaction,then the actual send is done asynchronously. That means the thread that issending the message does not wait for the ack from the broker. It’s thetransport thread that receives the broker’s ack and deals with it. Using anasync send in the case of running inside a transaction is done for performancereasons. By not waiting for the broker ack, you can send messages more quicklyto the broker. Until the transaction finally commits the messages received bythe broker will not be placed onto the brokers queue.

Putting this together the following happened in my test:

- The producer started a new transaction. This informationwas sent to the broker, the broker accepted the new transaction.

- The producer then sent a message to the broker.As this send was within an existing transaction it was done async. The call tosend the message returned immediately after sending the message. It did notwait for the broker’s ack.

- The broker did no accept this message but raiseda "javax.jms.ResourceAllocationException: Persistent store is Full, 100%of 67108864" back to the producer.

- The producers transport thread received theexception but didn’t know how to handle it. Instead it printed the followingdebug log message:
DEBUG - ActiveMQConnection - Async exception with no exception listener:javax.jms.ResourceAllocationException: Persistent store is Full, 100% of67108864. Stopping producer (ID:Mac.local-51375-1320685288867-2:1:1:1) toprevent flooding queue://TEST.IN.
It flagged the fact thatthere was an error but it did not mark the current transaction to be rolledback. This would have been the job of an exception listener, but none wasregistered.

- Theproducer then committed the transaction. From the producers point of view noerrors did happen (it never became aware of the JMSException) so it asked thebroker to commit the transaction.

- Thebroker committed the transaction just fine although the message did not getstored on the broker’s queue. From the brokers point of view all went fine. Theproducer did start a new transaction, it then sent some msgs, which the brokerrejected and then the broker got asked to commit the transaction. All fine fromthe brokers point of view. It’s the producer’s task to deal with theJMSException and to decide whether to rollback or not.

- Themessage(s) that got sent inside the transaction is (are) lost!

It does not necessarily require anActiveMQ broker that has reached its configured limits to run into thisproblem. If the transport for any reason fails to send the msg to the brokerbut works again on the transaction commit, then the same problem could arise. Thisis less likely to occur though. Transport related problems (e.g. connectionloss) will most of the time affect the ability to commit the transaction aswell (and not only the individual message send). When there is an exceptionduring commit, it will correctly roll back the entire transaction.

You wouldthink that using JMS Transactions would guard you against message loss. Ingeneral that is correct. However you need to make sure that any errors that canoccur within the transaction are being dealt with. In this scenario, there wasno exception listener registered in the producer that would deal with theJMSException. So I lost messages because I did not deal with the exceptionreturned from the async send.

When using Camel as a messageproducer, you currently cannot easily register an exception listener, whichthen ensures the transaction gets rolled back in case of any problems.

Luckily there are some simple solutions to this problem:

THE SOLUTION

.. is to either

1) Question if you really need to send your messages insidea transaction. When sending only one persistent message at a time inside atransaction, there is not much benefit over sending the message without atransaction. Persistent messages are sent synchronously by default so the producerwaits for the brokers ack before proceeding and it can deal directly with anyerror returned from the broker. Transactions are very useful though whensending multiple messages that need to all succeed or fail as one atomicoperation.

2) If you need to use transactions for sending your messagesto the broker, then configure the ConnectionFactory to always send messagessynchronously rather than using the default async mode. Changing to sync sendwill have a small performance impact, particularly when sending a batch of messageswithin one transaction. This performance impact however is probably negligiblein most scenarios.

You can switch to using a sync send by using thealwaysSyncSend=true property on the ActiveMQConnectionFactor, e.g. for a Springconfiguration:

</bean>

Alternatively, set jms.useAsyncSend=false on the broker URL,e.g.

tcp://localhost:61616?jms.useAsyncSend=false

3) Register a JMS exception listener programmatically sothat it reacts on any exceptions thrown by the broker in case of sending the messageasync. For a plain Java JMS client you can use

ActiveMQConnectionFactory connectionFactory =

new ActiveMQConnectionFactory(brokerUrl); connectionFactory.setExceptionListener(new ExceptionListener() {

public void onException(JMSException ex) {
// handle the exception
}
});

You can also set this exception listener on an existing JMSConnection object.

The somewhat tricky part may be to figure out whichtransaction needs to be rolled back, in case of having multiple threads sendingmessages concurrently to the broker.

I already mentioned this option is not easily done in casewhere the producer is a Camel route. We hope to improve Camel to have an out ofthe box exception listener registered in future. Till then CAMEL-4616

captures this request.

If you want to be on the safe side, the go with option #2and ensure a sync send of your messages.

ActiveMQ Network Bridge to Master/Slave broker pair

2011-09-09T11:01:00.000+02:00

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"/>

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

2011-07-26T09:37:00.007+02:00

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

2011-06-01T10:12:00.005+02:00

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

2011-03-23T13:46:00.002+01:00

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?

2011-03-10T10:35:00.006+01:00

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.

2011-02-24T17:45:00.003+01:00

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

2011-02-23T19:50:00.003+01:00

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?

2011-01-21T16:12:00.002+01:00

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!

2011-01-11T15:22:00.010+01:00

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

2010-12-27T16:40:00.005+01:00

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

2010-12-07T11:15:00.003+01:00

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

2009-12-17T17:33:00.005+01:00

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?

2009-11-04T10:22:00.006+01:00

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

2009-05-13T15:31:00.010+02:00

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?

2009-04-16T14:48:00.011+02:00

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.

Using the Camel Aggregator correctly

2009-01-23T15:07:00.012+01:00

I was recently working on a project that required the use of a Camel aggregator. Since I had not used the aggregator much before I started by reading through the Camel Aggregator documentation. Some questions remained though.
In my use case I needed to aggregate every five incoming messages of the same format together into a new exchange. Sounds very simple but I did not find it that straight forward. And that's the motivation for this post.
To have an example at hand; let's say I want to aggregate messages of the following general XML format (with different values of course):

<artist genre='Alternative'>
<name>The White Stripes</name>
<album>Elephant</album>
<album>Get Behind Me Satan</album>
<album>White Blood Cells</album>
</artist>

Suppose that's the message format of the incoming messages. I will build the aggregator according to this format.
As mentioned in the Camel documentation, each aggregator is configured by a correlation expression and an aggregation strategy.
The correlation expression is used to define which messages should be aggregated together. When the correlation expression is being evaluated at runtime on a particular message, it returns a value that is used as the correlation id under which messages are being aggregated. All messages that evaluate to the same correlation id are aggregated together. A mistake I made initially was assuming that all messages that do not match my correlation expression will not be aggregated. But every correlation expression will always evaluate to a particular correlation id for every message. Suppose I want to aggregate all those messages that have an alternative genre (genre='Alternative') and I define my aggregation strategy accordingly (we will explore that later). All messages that have a different genre set will still be aggregated somehow, depending on what id the correlation expression evaluates to for these messages at runtime. We will see more detailed examples of that later, but this is important to keep in mind.
The aggregation strategy on the other hand contains the logic for combining all message exchanges into a single aggregated message. The default aggregation strategy is org.apache.camel.processor.aggregate.UseLatestAggregationStrategy, which only aggregates the latest incoming message, discarding older messages. This is great for throttling purposes but did not serve my use case.

Generally a Camel aggregator is set up in Java DSL as follows (I will only use Java examples, XML based configuration will be quite similar):

from("somewhere").aggregate("somehow").to("some destination");

Let's not care too much about the from(…) and to(…) part of this route, they will be defined by your business use case. It is the aggregate(…) part that we are interested in here.
As mentioned above the aggregator is configured via the correlation expression (which messages to aggregate together) and the aggregation strategy (how to aggregate messages). So it comes as no surprise that the aggregate() method can take the following parameters:

public AggregatorType aggregate(Expression correlationExpression, AggregationStrategy strategy);

When defining a correlation expression, one can either correlate on a particular message header or on the message content. Pretty much all examples I have seen so far did correlate messages on the existence of a particular message header element. In Camel headers are named and can take values - they are basically (name, value) pairs.
Suppose some of my messages also have a header called "MusicCollection" with a value "Yes" set. I can then aggregate my messages based on this header and define my correlation expression as follows:

.aggregate(header("MusicCollection"), …).

This correlation expression will evaluate to the value of the header "MusicCollection" at runtime (the String "Yes" in the above example). If no such header is set, the correlation expression will evaluate to null at runtime, which can still be used as a key for aggregating messages.
But what If I don't want to or cannot use message headers? Perhaps the message was received from an external system without any headers. I can then also correlate messages based on their message payload as well. Using an XPath expression I can test if a specific element or structure is present in the message payload. In my example the messages start with as the root element, so it seems natural to define a correlation expression like this:

.aggregate(xpath("/artist", …).

This however won't work as expected. Internally the aggregator will use the Camel XPathBuilder to evaluate the XPath expression and the above expression will return a new instance of org.w3c.dom.NodeList for every new message. The aggregation however can only occur on a correlation expression that evaluates to the same object or primitive value at runtime for all messages to be aggregated (as that object or primitive value is used internally as a HashMap key for storing the aggregated messages into exchanges). So we need to define an XPath expression that evaluates to the same object (or key) at runtime. The trick is to select a proper XPath expression and to cast the evaluation result into a type like java.lang.String, such as in the following example:


.aggregate(…).xpath("name(/artist)='artist'", String.class).

I had to move the XPath expression after the aggregate() method call because of the additional String.class argument. The above example will take the name of the root element of my message (which will be "artist") and test if it equals the string "artist". For the sample message shown in the beginning, this expression will evaluate to the String "true". For any other XML messages that don't start with , this XPath expression will evaluate to the String value "false" and these messages will still be aggregated according to my aggregation strategy.
Aggregating on only the root XML element does not seem very useful. I might want to aggregate on the genre attribute of the element instead. Here is the corresponding correlation expression:

.aggregate(…).xpath("/artist/@genre", String.class).

This does serve my use case. If I want to aggregate all messages with an "Alternative" genre I could use:

.aggregate(…).xpath("name(/artist[@genre='Alternative'])", String.class).

Notice this will aggregate all messages starting with together into a new exchange. All other messages that have a different genre will also be aggregated together into a different exchange, no matter what their genre value is. If you wondered why I use the XPath function name() here, then that is because using name() will return the String "artist" as the correlation key for the example message above (the name of the XML element that matches my XPath expression). If I omitted the name() function, it would return a string representation of the node set itself. In the case of my example message the correlation key would be the string "White Stripes Elephant Get Behind Me Satan White Blood Cells". Using "artist" as the key seems better somehow.
If you are unsure to what key your own correlation expression evaluates at runtime, place a breakpoint into method org.apache.camel.processor.aggregate.DefaultAggregationCollection.add() and debug your Camel route from there. Check the value of the computed correlationKey object.
You can of course define your correlation expression in different expression languages than XPath such as XQuery, SQL and others.

That should cover the correlation expression part of the aggregator. The second part deals with the aggregation strategy. Remember, the strategy defines how messages are aggregated together. The default strategy used by Camel is the UseLatestAggregationStrategy, which out of x received messages only takes the latest message. If however I want to aggregate multiple messages together into a single message, I need to write my own aggregation strategy. This isn't a big deal and the Camel aggregator documentation shows an example implementation. I have re-used that code for my example as well.

public class MyAggregationStrategy implements AggregationStrategy {
public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
Message newIn = newExchange.getIn();
String oldBody = oldExchange.getIn().getBody(String.class);
String newBody = newIn.getBody(String.class);
newIn.setBody(oldBody + newBody);
return newExchange;
}
}

In this implementation we extract the messages of both Exchanges old and new, then simply concatenate the two messages together and place the result on the new Exchange. Nothing fancy here. We can then instantiate this class and pass its reference into the aggregator:

.aggregate(…, new MyAggregationStrategy()).

So we are done, right? Well, what we have not defined yet is a criterion as to when the aggregation is done. There are again multiple ways of doing so. If I want to aggregate a fixed number of incoming messages, I can easily use the batch options:

.aggregate(new MyAggregationStrategy()).
xpath("("/artist/@genre", String.class).
batchSize(5).
to(…);

In this example the aggregation will finish whenever it has processed 5 incoming messages no matter as to how they got aggregated (into which exchanges). Or I could specify a timeout after which the aggregated message is sent, irrespective of how many message have been aggregated:

.aggregate(new MyAggregationStrategy()).
xpath("("/artist/@genre", String.class).
batchTimeout(5000L).
to(…);

If I want to finish the aggregation when there are 5 messages aggregated together into one new exchange, then a more complex logic is required, which I can specify using the completedPredicate() expression:

.aggregate(new MyAggregationStrategy()).
xpath("("/artist/@genre", String.class).
completedPredicate(header("aggregated").isEqualTo(5)).

The example above assumes that there is a header called "aggregated" set and whenever its value reaches the number 5, the aggregation will finish. Of course some application level code needs to set this header and also increment the counter. This would typically be done in the custom aggregation strategy implementation, where increasing that number can be based on any application logic (see attached demo, which illustrates this approach).
I can also call a particular method on a Spring bean for deciding when to finish the aggregation by using a BeanExpression:

.aggregate(new MyAggregationStrategy()).
xpath("("/artist/@genre", String.class).
completedPredicate(
new BeanExpression("aggregatorStrategy", "isCompleted"))
).
to(...)

This requires a Spring bean with name "aggregatorStrategy" being defined in the Camel Spring configuration (your camel-context.xml) which exposes a public method called isCompleted().

<bean id="aggregatorStrategy"
class="com.progress.cs.camelaggregatortest.MyAggregationStrategy">

Check the attached demo for an example.
Finally, the entire aggregator code in Java DSL for my use case now reads:

from("…").
aggregate(new MyAggregationStrategy()).
xpath("/artist/@genre", String.class).
batchSize(5).
to("…");

Using a Camel aggregator can appear rather complex at first. But once you get to see some working examples it becomes much clearer (as with all things in life). I personally had some troubles getting the aggregator to work with an XPath based correlation expression. That part I wanted to share most of all in this article.

And finally here is the demo for this article. After extracting, follow the steps in README.txt to run it quickly.

WS-Security woes in ServiceMix 3.2

2008-07-29T17:09:00.003+02:00

Previously I had to get WS-Security UsernameToken based authentication working in ServiceMix 3.2 for a customer. All the customer wanted was to have an external SOAP client connecting securely to a service deployed in ServiceMix and have the client being authenticated and authorized by ServiceMix. Sounds like a straight forward use-case and the fact that there already is a ws-security demo in ServiceMix made me hope this would be a simple project.
As Murphy's Law dictates, it turned out to be the opposite.

My initial hope was to simply define a CXF-BC consumer that is WS-Security enabled and that authenticates the client and passes on the request to a CXF-SE service. So overall use-case is
External client -> CXF-BC consumer -> CXF-SE service.
This can be set up using Maven in less than 10 minutes. The tricky part was to configure security. The ws-security demo of ServiceMix already shows the configuration for using XML-Signature and XML-Encryption, so I could simply copy that part. For using WS-Security functions ServiceMix simply leverages WSS4J, so it becomes a matter of correctly configuring WSS4J interceptors for the incoming request and outgoing reply.
Adding WS-Security UsernameToken based authentication configuration to the WSS4J interceptor is relatively simple too:


<bean class="org.apache.cxf.ws.security.wss4j.WSS4JInInterceptor"
  id="TimestampSignEncrypt_Request">
  <constructor-arg>
    <map>
      <entry key="action" value="Timestamp Signature Encrypt UsernameToken"/>
      <entry key="signaturePropFile" value="alice.properties"/>
      <entry key="decryptionPropFile" value="bob.properties"/>
      <entry key="passwordCallbackClass" 
     value="org.apache.servicemix.samples.cxf_ws_security_usernametoken.KeystorePasswordCallback"/>
    </map>
  </constructor-arg>
</bean>

I simply had to add UsernameToken to the <action> element in the above WSS4JInInterceptor configuration (config for XML-Signature and XML-Encryption was directly taken from ws-security demo). I won't care about the client configuration here, as this could be any SOAP client but need to add that my client was configured to transmit username "alice" with password "password".
For authorization to work, I had to add these credentials to conf/users-passwords.properties


# users-passwords.properties
smx=smx
alice=password

and make alice a member of the administrators role in conf/groups.properties (for a real use-case I would have defined a new role):


# groups.properties
admin=smx,alice

Finally to restrict only users of the administration role to invoke my service, I had to change conf/security.xml


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

Easy I thought and went to testing it. But it did not work. There was simply no authentication happening. Only when debugging through the code I realized the CXF-BC component did not delegate these credentials to JAAS for authentication! So there was no authentication possible in CXF-BC! I raised a bug report and took a deep breath...
The good news is that by now this bug got resolved and things work as expected. However you need a very recent version of ServiceMix to have this fix included. Now if a client sends a WS-Security header containing username and password, these will be authenticated correctly by JAAS inside ServiceMix.
Still at the time of discovering the bug I had to get things working for the customer and got to know that the HTTB-BC component supports username/password based authentication. So I started to replace my CXF-BC consumer with a HTTP-BC consumer. And this is where the real pain started.
The configuration again is straight forward and does not look much different from the CXF-BC consumer configuration. However, for authentication to work in HTTP-BC, I must set soap="true" in my bc consumer config:


<beans xmlns:http="http://servicemix.apache.org/http/1.0"
       xmlns:greeter="http://apache.org/hello_world_soap_http"
       xmlns:soap="http://servicemix.apache.org/soap/1.0">
  <http:endpoint service="greeter:SOAPServiceWSSecurityHTTP"
    endpoint="endpoint"
    role="consumer"
    targetService="greeter:SOAPServiceWSSecurity"
    targetEndpoint="endpoint"
    locationURI="http://localhost:9000/"
    defaultMep="http://www.w3.org/2004/08/wsdl/in-out"
    soap="true">
    <http:policies>
      <soap:ws-addressing />
      <soap:ws-security receiveAction="UsernameToken" keystore="default" />
    </http:policies>
  </http:endpoint>
</beans>

This is because the WSSecurityHandler that is called by the HTTP-BC consumer expects a DOM representation of the SOAP header, which won't get created unless soap="true" is set. With that I got the HTTP-BC component to authenticate the client, but the request failed to be un-marshaled by the CXF-SE runtime. Setting soap="true" in my http consumer had the side effect of stripping off the SOAP header from the request before sending it on to the next component inside ServiceMix. The CXF-SE runtime however requires a valid SOAP request, which it now did not get. Aargh...

So what to do with this dilemma? The easiest option I could think of was to insert a Camel component between the HTTP-BC consumer and the CXF-SE service that adds a default SOAP envelope to the In message using xslt transformation and removes the SOAP envelope from the Out message produced by the CXF-SE component again (as the http-bc component will add an envelope when soap="true" is set.) Not an ideal solution and rather a hack than a proper workaround. Another option would have been to use a custom CXF interceptor that takes care of adding/stripping the SOAP envelope but I felt a Camel XSLT component will be simpler to use.
I inserted the Camel component that did add/remove the SOAP envelope. Still I had a problem with the reply. The reply message had to be signed and encrypted but that was not the case, despite my configuration being correct. After another few hours of debugging (I am new to the codebase) I realized this part had yet not been implemented and I had hit another bug. Because of that bug outgoing SOAP reply message cannot be encrypted nor signed when using the HTTP-BC component.

I still found some other issues on the way that I raised accordingly. Password digests as defined by the WS-Security UsernameToken spec are currently not supported (passwords need to be send in clear text in the SOAP header - but the header can still be encrypted, so this is no security problem). And the call to authorization in ServiceMix does not produce any logging. Finally I raised a few documentation issues.

Lessons learned:
As you can see I hit a number of issues on the way. But the good news is that these will now be addressed and fixed in future versions of ServiceMix, making the product more stable and easier to use. The most important bug is already resolved so there is no need to use a HTTP-BC consumer anymore when doing WS-Security (and I highly recommend against it).

The demo that I worked on is already incorporated into the cxf-ws-security demo of ServiceMix in the current release.

The most difficult part for me was to hit very low level error messages that did not tell me much about the real cause of the problem. This can be quite frustrating and the only way I got to understand the error was to debug through the source code. So error reporting needs a hell lot of improvement in ServiceMix.

Secondly, the low-level XML configuration is highly error-prone. ServiceMix really needs proper tooling that free the developer from configuring xml files. It is not only easy to make mistakes in the config, there are also so many configuration options that have implications unknown to the regular developer (like the soap="true" story).

... the namespace on the "message" element, is not a valid SOAP version.

2008-05-28T09:19:00.005+02:00

Recently I was working on a SMX usecase:
external client -> CXF BC consumer -> CXF SE

All configured nicely it still failed at runtime with


INFO: Interceptor has thrown exception, 
unwinding now org.apache.cxf.binding.soap.SoapFault: 
"http://java.sun.com/xml/ns/jbi/wsdl-11-wrapper", 
the namespace on the "message" element, is not a valid SOAP version.

It took me a while to figure out what's wrong. In the end I only had to set useJBIWrapper="true" in the config of my CXF SE component:


<beans xmlns:cxfse="http://servicemix.apache.org/cxfse/1.0">
  <cxfse:endpoint useJBIWrapper="true">
    <cxfse:pojo>
      <bean class="org.apache.servicemix.samples.cxf_ws_security.GreeterImpl" />
    </cxfse:pojo>
  </cxfse:endpoint>
</beans>

Those error messages really need to get improved in SMX. There are too many low level errors raised that give you no indication of what the real problem is.

Chronic - a date and time parser for Ruby

2008-04-15T10:33:00.002+02:00

I have recently come across Chronic, a very elegant Date and Time parser library for Ruby. It can be easily installed via RubyGems and can process all sorts of dates and times written in natural language formats. If you need to parse date in Ruby this might be the library you have been looking for.

There is an interesting post on parsing US and non-US date formats in this post.

Kerberos authentication in Java

2008-04-07T12:32:00.005+02:00

Have you used Kerberos authentication in Java using JGSS? The GSSContext interface defines the two methods initSecContext() and acceptSecContext() for an initiator and acceptor of a Kerberos authentication request. JGSS is configured in a file typically called jaas.conf, where you specify the Kerberos keytab files, Kerberos principal name and other properties that configure the authentication process.

According to the Kerberos authentication algorithm, only the initiator (or client) needs to contact the Kerberos authentication server, asking for a service ticket. The service ticket is later passed on to the server that the client wants to authenticate against and it contains all the information that the server (or acceptor) requires for authenticating the client. That implies the server does not need to communicate with the Kerberos authentication server when authenticating a client, as it is emphasized in pretty much every description of the Kerberos protocol.

Still when using JGSS you might have noticed that your service does contact the Kerberos authentication server as part of the authentication process. The server will request a ticket granting ticket from the KDC even though this is not needed. This behavior has caused me some headaches in the past as I did not understand why the server would need to contact the KDC.

There is a not very much documented property in jaas.conf that will solve this issue. If in your com.sun.security.jgss.accept stanza of jaas.conf you specify isInitiator=false, then the server will not request a TGT. Officially this property is only supported from Java SE 6 b89 onwards, however I verified that the later patches of Java 5 also include it.
So when setting up jaas.conf, I recommend adding isInitiator=false to the configuration of the acceptor.

This topic is also discussed here.

Let's change the default transport protocol used in ServiceMix 3

2008-04-04T14:40:00.002+02:00

Today I want to post about a potential issue that I came across recently when running tests using the FUSE ESB (which is IONA's distribution of Apache ServiceMix).

Currently ServiceMix 3.2 configures the underlying ActiveMQ broker to simply use tcp as the transport protocol. This is defined in conf/servicemix.properties:

activemq.url = tcp://${activemq.host}:${activemq.port}

While this configuration works fine and is sufficient for using ServiceMix in development, it will not recover from any network failures and is therefore not suitable for running in a production system. In case of a network failure or connection inactivity timeout, ActiveMQ will simply raise an exception but won't try to reconnect. You will need to restart ServiceMix in order to recover from such error. For automatic reconnection to happen, the failover protocol should be used instead. Hence the default configuration should be changed to

activemq.url = failover://(tcp://${activemq.host}:${activemq.port})

The failover protocol should really be the default protocol and enabled out of the box. I logged SM-1302.

Bug in gdm under SuSE 10.3 (and perhaps in other Linux distributions)

2008-02-29T11:36:00.001+01:00

Anyone who is running Linux with GNOME in VMWare might have come across a rather serious problem with gdm lately after doing an upgrade of the system with the latest patches. I am running OpenSuSE 10.3 and installed the latest patches just a few days ago. The patch installation went fine but next morning when I tried to reboot my Linux image, gdm failed to start up. Gdm initially started up but then crashed just before or shortly after showing the login prompt. When gdm was started the first time it ran a few seconds longer before crashing. You won't even have enough time to login. On subsequent restarts, GDM crashed almost instantly for every restart attempt. It attempted 6 restarts, then raised the following error to the console, slept for 2 minutes and started over again:

"The display server has been shut down about 6 times in the last 90 seconds, it is likely that something bad is going on. Waiting for 2 minutes before trying again on display:0"

The log in /var/logs/Xorg.0.log should had the following lines at the end:
Backtrace:
0: /usr/bin/X(xf86SigHandler+0x81) [0x80e6d81]
1: [0xffffe420]
2: /usr/bin/X [0x817b185]
3: /usr/bin/X(CompositePicture+0x150) [0x8161be0]
4: /usr/bin/X [0x8167a1f]
5: /usr/bin/X [0x8164d75]
6: /usr/bin/X [0x815809e]
7: /usr/bin/X(Dispatch+0x1af) [0x808f68f]
8: /usr/bin/X(main+0x47e) [0x807717e]
9: /lib/libc.so.6(__libc_start_main+0xe0) [0xb7d1ffe0]
10: /usr/bin/X(FontFileCompleteXLFD+0x1e5) [0x8076501]

Fatal server error:
Caught signal 11. Server aborting

After various tries I was still unable to resolve the problem but found a way to bring up the X Windows system. I had to login as root from console and run gdm -stop. The next time gdm was restarted (after the 2 min had elapsed) it did come up fine.

So I went on the web to see if someone else had encountered the same problem as well. When I checked SuSE's bug database, I realized I was not the first one to encounter this issue. A bug was already raised https://bugzilla.novell.com/show_bug.cgi?id=350318.

Apparently the culprit is a patch to the Control-Center2 (https://bugzilla.novell.com/show_bug.cgi?id=337434) that seens to break gdm.

This is quite a severe problem and only seems to affect Linux running inside VMWare. I am not yet sure if other distributions are effected as well, most likely they are.

simple Ruby SOAP client

2008-02-25T11:42:00.006+01:00

I am fairly new to Ruby and am quite amazed how simple and elegant this language can be. In particular I was surprised how simple it is to write a SOAP client in Ruby. Where in most programming languages even a simple SOAP client takes about two screenshots of code (if not more), a Ruby SOAP client can be boiled down to these few lines of code:

#soap_client.rb
require 'soap/wsdlDriver'

WSDL_URL="file://C:/hello_world.wsdl"

soap = SOAP::WSDLDriverFactory.new(WSDL_URL).create_rpc_driver("SOAPService", "SoapPort")

#invoke WSDL operation sayHi
result = soap.sayHi()
printf "Returned message reads: " + result + "\n"

#invoke WSDL operation greetMe
result = soap.greetMe("New Ruby User")
printf "Returned message reads: " + result + "\n"

In this code I am referring to a wsdl definition that defines two WSDL portType operations named sayHi() and greetMe() that both return a string.
In essence the code boils down to 4 lines of code that are of relevance. Of course this code is everything but object oriented and it is somewhat unfair to compare it against a properly implemented SOAP client written in other object oriented programming languages. Still this example is surprisingly simple and straight forward. I have used this code a number of times in order to quickly implement a SOAP test client. All you need to change is the name of the business method you actually want to invoke on (plus their arguments) and the location of the WSDL file. Thanks to Ruby you don't even need to recompile your changes.
Sure, most SOAP toolkits also provide a way to generate a fully working client and server without requiring you to write any code. Still the generated code is tied to the WSDL interface. Changes to one of the operation signatures typically require a regeneration of your test client/server.
Ruby rocks (and I am just starting to hit the tip of the iceberg)!