Rollback JMS transactions in MDB Vs MDP

If you are writing an application which has a message driven bean or a message driven POJO you might want to put back the message in JMS destination if there is an exception in your application logic. In other words rollback the JMS transaction if there is an exception in onMessage. On a side note the acknowledgement mode of the message is considered to be AUTO_ACKNOWLEDGE instead of CLIENT_ACKNOWLEDGE as in the AUTO mode the messages are automatically removed from the queue unless a rollback is issue. Let us discuss how to issue a rollback in the case of an MDB first and then in a MDP.

Issuing a rollback in a Message Driven Bean (MDB)

A sample MDB implements MessageDrivenBean and MessageListener as follows

public class ExampleMDB implements javax.ejb.MessageDrivenBean, javax.jms.MessageListener{

javax.ejb.MessageDrivenContext context;

    public void setMessageDrivenContext(javax.ejb.MessageDrivenContext aContext) {
        context = aContext;
    }
// ejbCreate and ejbRemove methods go hear.

// onMessage needs to be implemented here
public void onMessage(javax.jms.Message inMessage) {

  // do your business logic here
 // if error in business logic rollback by calling
 context.setRollbackOnly();


}

}

The message driven context is set by the container in a container managed bean. The rollback is achieved by calling the context.setRollbackOnly() when this happens the message is put back in the queue. One should be carefull as the onMessage will be called repeatedly until the processing is fine and the rollback is not issued.

Issuing a rollback in a Message Driven POJO (MDP)

In a Spring MDP the task of issuing a rollback is as straightforward as just throwing a RunTimeException.

public class ExampleMDP implements javax.jms.MessageListener{


public void onMessage(Message message) throws RuntimeException{

   // if error in business logic throw a RuntimeException
}
}

When the MDP is configured in the application context xml file make sure it is sessionTransacted, by setting it's value to true. Refer to my earlier article
how-to-write-message-driven-pojo

IllegalStateException in Spring's JmsTemplate

I have been using Spring's JmsTemplate to send messages to Queue's and Topic's in SUN Java Message Broker. The code to implement this was really simple you just configure a bean in your context called jmsTemplate with points to class org.springframework.jms.core.JmsTemplate, autowire this bean in your code and there you have all the methods to send all kinds of objects to your JMS destination.

Every thing worked fine, but now and then the jmsTemplate.convertAndSend method used to throw an org.springframework.jms.IllegalStateException: Session closed; nested exception is javax.jms.IllegalStateException: Session closed. Now I am not sure if this is very specific to my version of JMS broker, but some more reading in the spring forum showed that it happens in other JMS brokers aswell. There seems to be a bug with the JmsTemplate spring code especially when it is multithreaded. It seems that the JMS session is shared across multiple threads and at times one thread gets hold of a closed JMS session.

I had to find a way around this issue till the bug is fixed by handling the exception and using a custom boilerplate code for sending JMS messages,which I agree sucks big time.

How to write a message driven pojo (MDP)

MDP is very simple to write in spring. There are quite a few examples out there. Follow these simple steps

Step1: Write a java class which implements javax.jms.MessageListener interface and implement the onMessage method

public class MyMessageListener  implements javax.jms.MessageListener{


    public void onMessage(Message message) throws RuntimeException{
 try{
  // do what you have to here
 }catch(Throwable ex) {
           throw new RuntimeException(ex);
        }
    }
}

Step2: Add the bean to your spring context.
In your spring context XML file that you have defined add the following entry. if you are using anotation then this is optional

<bean id="myListener" class="MyMessageListener"/>

Step3: Add a jms connection factory
Connection factories can been directly defined in the spring context in which case you need to also define the connection pooling. If your application is going to reside in an
application server then you can use JNDI to get the connection factory. If you are using the second approach you define the jndiTemplate then the connection factory

<bean id="jndiTemplate" class="org.springframework.jndi.JndiTemplate"/>
<bean id="jmsQueueConnectionFactory"  class="org.springframework.jndi.JndiObjectFactoryBean">
 <property name="jndiTemplate">
 <ref bean="jndiTemplate"/>
 </property>
 <property name="jndiName">
 <value>jms/QueueConnectionFactory</value>
 </property>
</bean>

Where jms/QueueConnectionFactory is defined as a resouce in the application server along with connection pool settings.

Step4: Add the jms container to your spring context XML file.
This step attaches your POJO to a specific JMS resource (Queue/Topic)

    <bean id="myJmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="jmsQueueConnectionFactory"/>
        <property name="destinationName" value="MyExampleQ"/>
        <property name="sessionTransacted" value="false"/>
        <property name="messageListener" ref="myListener" />
        <property name="concurrentConsumers" value="1" />
        <property name="receiveTimeout" value="30000" />
    </bean>

And there you have it a simple message driven POJO called myListener mapping to the class MyMessageListener. It listens to messages in the queue named MyExampleQ
this is mentioned in the property destinationName. In this example sessionTransacted is set to false which means that if an exception occurs in onMessage method
the message will not be rolledback i.e. the message will be removed from the queue even if onMessage method failed. In most cases you might want the message to remain
in the queue if your operations in onMessage failed, in which case you need to set the value to true.

In this example concurrentConsumers is set to one you can use this property to better your performance by increasing this value.

Using Client Acknowledge mode to expose a JMS topic as a Web service

A JMS session has three acknowledge modes AUTO, CLIENT and DUPS_OK, messages are removed from the JMS destination (Queue or Topic) only when the session has acknowledged that the client has received the message. In AUTO acknowledge mode the message is acknowledged as soon as the receive method has been completed. In DUPS_OK acknowledge the message is acknowledged lazily which means the message could be sitting in the destination for some time before the broker removes it at its will. In CLIENT acknowledge mode the client after receiving the message acknowledges it by calling its acknowledge method.This mode is key when you want to expose your JMS destination as a web service.

The one problem that needs to be solved when exposing a JMS destination as a web service is reliability. How do we make sure that the web service client has successfully received the message. In other words how do we support distributed transaction. The following example demonstrates this.

Target is to expose a JMS topic called yahoonews as web service.

Step1. Create a durable subscriber with CLIENT acknowledge mode

public Class DurableSubscriber{
  Context jndiContext = null;
  TopicConnectionFactory topicConnectionFactory = null;
  TopicConnection topicConnection = null;
  TopicSession topicSession = null;
  BasicTopic topic = null;
  TopicSubscriber topicSubscriber = null;

  public DurableSubscriber(){
      try {
          topicConnectionFactory =  new com.sun.messaging.TopicConnectionFactory();
          topicConnectionFactory.setProperty("imqBrokerHostName",brokerHost);
          topicConnectionFactory.setProperty("imqBrokerHostPort",brokerPort);         
       
          topicConnection = (com.sun.messaging.jms.TopicConnection)
          topicConnectionFactory.createTopicConnection();
          topicConnection.setClientID("yahoouserid1001");

          topicSession =
          topicConnection.createTopicSession(false,
          Session.CLIENT_ACKNOWLEDGE);
          topic = (com.sun.messaging.BasicTopic)topicSession.createTopic("yahoonews");
      } catch (Exception e) {
          e.printStackTrace();
          if (topicConnection != null) {
              try {
                  topicConnection.close();
              } catch (javax.jms.JMSException ee) {}
          }
      }
  }// end constructor

  public void startSubscriber(){
      try {

          topicConnection.stop();
          topicSubscriber =
          topicSession.createDurableSubscriber(topic,
          "finanancialnewsSubscriber");
       
          topicConnection.start();
      } catch (javax.jms.JMSException e) {
          System.err.println("Exception occurred: " +
          e.toString());
      }

  }

  public javax.jms.Message getMessage(int timeout) throws Exception{
  
     javax.jms.Message m = topicSubscriber.receive(timeout);
     return m;
  }
}

Step2. Create a wrapper class that will exposed as web service using REST or SOAP.
The web service should have two functions getMessage and ackMessage.
The class should also store the message that needs to be acknowledged.

public class JMSWebService{

  private static DurableSubscriber ds = new DurableSubscriber();
static{
   ds.startSubscriber();
}
  private javax.jms.Message ackMessage;

  public String getMessage(){
      if (ackMessage==null)
          ackMessage = ds.getMessage(1000); // 1 sec blocking call to get messages
      String str = ackMessage.getText();
      return str;
  }

  public void ackMessage(){
      if (ackMessage!=null)
          ackMessage.acknowledge();
      ackMessage = null;
  }
}

In this example reliability is achieved by having the ackMessage service along with getMessage, the client has to call the ackMessage function to indicate that all is well at the client side and that the client has commited the transaction at his end. If the ackMessage call is never made and the client keeps calling the getMessage function the transaction will not be commited and so the same message is returned to the client.