We’re currently working on a JMS-drived application, which is being used as an integration point between several systems. We’ve defined a standard exception handling process – checked exceptions for those errors that can be handled by the system, unchecked exceptions for those errors that cannot be handled. The unchecked exceptions are all allowed to bubble up, where they are caught at the top level (and the top level only).
Given that the entire system is driven by received messages, the “top” of our application is our MessageListener, which before today looked a little like this:
public void onMessage(Message m) {
try {
process(m);
} catch (Throwable t) {
log(t);
}
}
As per our exception handling strategy, this is the only place where unchecked exceptions are caught. Today we started implementing the various sad-path scenarios, the first of which was “what happens if processing the message fails?”. In our case this scenario translates as process throwing an unchecked exception. This being our first scenario, we’ve assumed that any runtime exception is a transient error – possibly down to one of the systems we’re integrating with being down. As such we decided that we’ll want to attempt to process the message again.
The simplest way to re-try a message is to use a transactional JMS session and rollback the session, as this returns the message back on to the topic/queue – we’d then specify a maximum number of times a message can be retried. It also follows that when using a transactional session you need to commit the session if the message was successfully processed.
Adding the code to commit a transaction is straightforward – but we do have to expose the session inside the listener (we’re using a topic here for monitoring purposes):
public OurMessageListener(TopicSession topicSession) {
this.topicSession = topicSession;
}
public void onMessage(Message m) {
try {
process(m);
//if we get here, we've processed the message
topicSession.commit();
} catch (Throwable t) {
log(t);
}
}
Adding the code to roll a session back is a bit more work:
public void onMessage(Message m) {
try {
process(m);
topicSession.commit();
} catch (Throwable t) {
log(t);
topicSession.rollback();
}
}
Great so far, but our next “sad-path” scenario is going to give us a little more trouble. What if we receive a message that we can’t translate into something meaningful? We don’t want to re-try the message, as we know we’re not going to be able to handle it later – the message is just plain bad. To handle this case, we separated out the message processing and had it throw a checked exception:
public void onMessage(Message m) {
try {
Object command = new MessageHandler(m);
process(command);
topicSession.commit();
} catch (MessageInvalidException e) {
log(t);
//we don't want to retry the message
topicSession.commit();
} catch (Throwable t) {
log(t);
topicSession.rollback();
}
}
This works as far as it goes, but up to this point I’ve been oversimplifying things a little. We’d abstracted our use of JMS behind two simple methods, which so far had been good enough for use in all our tests, and both client and server code:
public interface MessagingFacade {
void subscribe(MessageListener listener);
void publish(String xmlToSend);
}
The subscribe call hides all the JMS plumbing – including the creation of the session itself. If we want to pass our session into the message listener, we need to expose it from the MessagingFacade or create it ourself – either way we kind of defeat the object of the facade. If we don’t use the facade, we end up complicating much of our code.
The solution we came up with was to create a TransactionalMessagingListener like so:
public class TransactionalMessagingListener
implements MessageListener {
public TransactionalMessagingListener(
TopicSession topicSession,
MessageListener delegate) {
...
}
public void onMessage(Message m) {
try {
delegate.onMessage(m)
topicSession.commit();
} catch (Throwable t) {
log(t);
topicSession.rollback();
}
}
Our underlying message listener is no longer the top of our system so doesn’t need to log throwable. Nor does it need to worry about the TopicSession, so becomes much simpler – we catch and log the checked exception related to message processing and let any unchecked exceptions bubble up to the TransactionalMessagingListener:
public void onMessage(Message m) {
try {
Object command = new MessageHandler(m);
process(command);
} catch (MessageInvalidException e) {
log(t);
}
}
And finally we change our MessagingFacade a little, making the subscribe method more specific by calling it subscribeWithTransaction, and wrapping the listener with our new TransactionalMessageListener:
public void subscribeWithTransaction(
MessageListener listener) {
...
TransactionalMessageListener txnListener =
new TransactionalMessageListener(
topicSession,
listener
);
...
}
And there we have it. All the code is simple and testable – and not a dynamic proxy in sight (take that AOP nut-cases!). I still can’t help thinking there was a simpler way of handling this though…
magpiebrain 
7 Responses to “JMS, transactions and exception handling”
How about – just writing MessageListener POJOs with no complex logic; letting any runtime exception thrown bubble up.
Then if you can detect an invalid message in some way, send it to a dead letter queue? Then for bad messages, you process them successfully but place them on a DLQ. For any bad errors, you let runtime exceptions bubble up.
Once you’ve done that you can use a JMS container (MDB container or a Message Driven POJO container, like the one for Spring) to process the messages and let it worry about handling JMS transactions start/commit/rollback, XA transactions or message acknowledgements, re-try, timeouts and the like?
i.e. keep the middleware stuff out of your business logic & focus on the main logic you need to do – process the messages and detect bad messages.
If you wanna Message Driven POJO container using Spring, try this…
http://activemq.codehaus.org/JCA+Container
Otherwise any old MDB container would do the trick (though you might wanna wrap your MessageListeners’s in MDB wrappers to make them full MDBs).
Well, I am keeping the complex (middleware) logic out of the code. What we end up with is a simple MessageListener implementation which just calls process(). Our TransactionalMessageListener delegates calls to our simple listener, and it handles the transactions. In reality, our simple (transactionally unaware) MessageListener processes the message (which in our case involves deserializing XML) and passes the resulting object into a POJO. So we have Transaction Wrapper->Message Umarshalling->Processing.
I’m not sure how using spring would do anything different, apart from perhaps having me code a configuration file rather than write some actual code. Oh, and I’d love to read the ActiveMQ documentation, but it seems all the example links go to your now non-existant CVS repository 🙂
And as for keeping transaction management out of business logic, how does that work where transaction management is business logic? For us, we have faliure scenarios which as an implementational detail have been implemented as transactions.
You can, if you like, write your own JMS consumption code (presumably handling JMS connection pooling & thread pooling?), setting up subscriptions, transaction handling with rollback & retry code – with or without XA – if you like.
But this is exactly what the JMS containers (be it MDB or MDP) do for you.
(Also note that a JMS container will do nice JMS connection, session & thread pooling for you too – not using a regular client side JMS consumer, but using the ServerSessionPool stuff – with a configurable thread pool so you can define how parallel things go & how threads work across multiple subscriptions etc).
This reminds me a bit of Joe’s recent post on DAO frameworks. You’re effectively writing a mini-JMS container to handle the transactions & retry. This is totally fine (I’ve done the same before in the past).
However you could try just reuse an MDB / MDP container, which does everything you need – plus lots more (connection pooling, thread pooling, nicer configuration, full XA support if required etc etc) – without using proprietary APIs.
i.e. there’s no reason why you can’t do this using JBoss, Geronimo or Spring today.
The only real trick is – for invalid messages, send to a DLQ, otherwise throw a runtime exception for system failures. Other than that you can reuse the rest – if you wish. But by all means write your own mini-JMS container if you wish – horses for courses etc.
(BTW the links work currently for the ActiveMQ examples on the JCA container 🙂
For us, we have a single entry point – this one message handler, so a mini-JMS container is fine – it doesn’t really warrant JBoss, Geronimo. I’m still unsure what Spring would give us in this scenario – if we use spring we’d have to configure transaction management anyway, and I’d rather write code (which is easier to test than config files). If/when our transaction management becomes more complicated (we might have to end up implementing two-phase commit or saga transactions) Spring might be a sensible choice.
Quote the Javadoc: “Although it is possible to have multiple QueueReceivers for the same queue, the JMS API does not define how messages are distributed between the QueueReceivers.”
If I’m not mistaken, this means that most of the benefits claimed by using J2EE containers or Spring for processing producer-consumer style messages is lost – meaning session and whatever polling & thread pool & stuff.
As for topics I think a thread pool is easily written. I see no reason whatsoever to tie myself to a particular framework and philosophy in order to achieve that stuff.
Come XA, I definitely need a transaction manager…
Radu – not really; that just means its up to a JMS provider to decide how to distribute messages across queue receives on the same queue. i.e. how it load balances.
Also note, doing efficient parallel consumption across threads in JMS is harder than it might first appear – quick answer, take a look at ServerSessionPool in JMS.
I kind of agree with James. Except that it’s very difficult to abstract JMS provider behaviour away from the logic b’cos the spec is so loose. JMS providers have too much freedom – actually calling JMS a spec is kind of a joke. Unless you decide to depend on a specific provider’s behaviour (ie make activemq/joram/jbossmq part of the architecture) it’s really tricky to tee up the SLAs required by the business to the relible behaviours of the provider (case in point: a transaction is not the same as an order fulfilment). Worst case is you will depend on a provider specific behaviour unbeknownst.
This problem bites as well when it comes to web based messaging (ie teeing up BizTalk exchanges with the business requirements around duplicates).
So, the last time I worked this kind of system (XML messaging over JMS), we went live and naturally the surprise exceptions came in. But what was really biting us were the unhappy side effects of either trying to recover inside onMessage() or the JMS provider sending the message back in. For the most we didn’t see transient failures at that level; if it didn’t go the first time it probably ain’t gonna go period. Web systems are very different in terms of transient failures (retrying is entirely justified).
So, after a few months I decided to change the onMessage() failure behaviour to use an an application level DLQ – a file system write. We removed a lot of code. No exceptions ever thrown back to the provider, no smart handling, no recovery behaviour. To heck with commits and rollbacks, just crash out, log the problem, plonk the message on disk, and wait for the next message. As dumb as you like, but with no partial failure. It’s unintuitive, but it was a very good way to get a highly robust application that had the required business semantics. What was lost in leveraging JMS infrastructure (after all, who the heck writes a custom DLQ on top of JMS), was gained in transparency.