Transactional behavior in JMS

[Former Member]

On WAS 6.40 I have a problem trying to get the following behavior when using JMS:

A message is sent from a session bean method (TX_REQUIRED) and then we mark the transaction for rollbackonly, i.e. aborting the transaction. Expected behaviour is that the message should not be sent. This works for a number of application servers however it fails on WebAS.

I am using XAQueueConnectionFactory configured in VisualAdmin. I tried several scenarios for session creation as well as using XA interfaces explicitely:

XAQueueConnectionFactory factory = (XAQueueConnectionFactory) ctx.lookup( JNDIUtils.Q_CONFAC );

XAQueueConnection _queueConnection = factory.createXAQueueConnection();

_queueConnection.createXAQueueSession();

So is there a way in netweaver to achieve this behavior without using explicit transactional attributes for sessions? We rely heavily on using container managed transactions.

Any help is appreciated!

[Former Member]

Hello Mikhail,

You are creating the JMS session with transacted flag false, because of this flag the session will not be enlisted in the EJB transaction.

As a workaround you can update your code to use a true flag :

_queueSession = queueConnection.createQueueSession( true, Session.AUTOACKNOWLEDGE );

Best Regards

Peter Peshev

[Former Member]

Hi Peter,

Actually that was the first thing that I tried. However it turns out that this approach can work only if I call explicit

commit()

rollback() 

on the session object.

If I just do this change JMS stucks - no messages are sent. In the code that I listed before I get a failure that no reply message was received within 5 seconds. So basically the message never comes to MDB listener, and that made me think that commit is never sent to the jms session that I want to participate in the container managed transaction.

Mikhail

[Former Member]

Hello Mikhail,

If you are creating the session with flag transacted=true than all manual explicit calls to commit() are supposed to end in TransactionInProgressException. It seems strange that those worked.

I am thinking of two possible reasons :

- The container managed transaction is not configured properly.

- The lookup string is not quite right. What is the value of JNDIUtils.Q_CONFAC ? Does it start with java:comp/env/jms/ ?

Can you please post the deployment descriptor, the value of JNDIUtils.Q_CONFAC and the file that created the jms factories (jms-factories.xml) ?

Best Regards

Peter Peshev

[Vlado]

Hi Mikhail,

In one of your postings above you said:

> A junit test sends two messages to a sendTestQueue through a call to a sessionbean method (transaction required).

However, in your first post you finished with:

> So is there a way in netweaver to achieve this behavior without using explicit transactional attributes for sessions?

So, assuming you have not declared transaction attribute for the session bean method, the default transaction attribute for session beans is Supports. Therefore there will be no global transaction and the effect of setRollbackOnly() will be none.

If this is not the case, please follow Peter's suggestions above.

Best regards,

Vladimir

[Former Member]

Hi Peter,

I did not set the transacted attribute to true - please check out the original code that I listed above - it's false. However I did try it and it did not work for obvious reasons that you have mentioned.

Lookups are not done through resource references but the first test (when sending a message and getting reply) works fine meaning that lookup worked ok. The only thing that fails is sending a message in a failing transaction (where we rely that the message should not be sent).

public static final String Q_CONFAC = "jmsfactory/default/GRQueueConnectionFactory";

GRQueueConnectionFactory is of XAQUEUECONNECTIONFACTORY type.

Here is an extract from jmsfactories.xml:

    <connection-factory>
      <factory-name>
        GRQueue
      </factory-name>
      <context-factory-type>
        <link-factory-name>
          jmsfactory/default/GRQueueConnectionFactory
        </link-factory-name>
        <initial-context-factory>
          com.sap.engine.services.jndi.InitialContextFactoryImpl
        </initial-context-factory>
        <provider-url>
        </provider-url>
        <security-principal>
        </security-principal>
        <security-credentials>
        </security-credentials>
      </context-factory-type>
    </connection-factory>

The session bean methods are configured as transaction required.

Best regards,

Mikhail

[Former Member]

Hi Vladimyr,

It is a good catch from your side and my bad wording. When I wrote "explicit attributes for sessions" I meant JMS sessions. Basically what I wanted to avoid is to explicitely commit jms sessions.

I used the following settings in ejb-jar:

       <container-transaction>
            <method>
                <ejb-name>JMSMessageApplication</ejb-name>
                <method-intf>Remote</method-intf>
                <method-name>*</method-name>
            </method>
            <trans-attribute>Required</trans-attribute>
        </container-transaction> 

I am relying that this gives enough indication to the container and I don't have to indicate transactional attributes for each method separately.

Thanks,

Mikhail

[Former Member]

Hi again,

I just noticed that you said that setting the flag to true should not allow to explicitely commit (my fault). Normally it works vice versa and the same is listed in SAP JMS samples for transacted sessions.

Regards,

Mikhail

[Former Member]

Hello Mikhail,

The problem with the lookup is that it works but it bypasses the standard JMS connector architecture. You are working directly with the JMS resource and there is no integration with the EJB transactions. The component that

should handle the transaction management between the JMS and the EJB is missing.

The quickest workaround should be to use the following lookup string :

public static final String Q_CONFAC = "jmsFactory/GRQueueConnectionFactory";

and use transaction=true in your example.

(Please make note of the capital F in jmsFactory.)

The recommended solution would be to lookup starting with java:comp/env...

Please have a look at the following link for more information, more examples and better explanations :

http://help.sap.com/saphelp_erp2004/helpdata/en/44/976986db254d16b2afae73f0edd8b1/frameset.htm

Best Regards

Peter Peshev

[Former Member]

Thanks a lot, Peter,

It solved it.

Regards,

Mikhail

[former_member199535]

Hi Mikhail,

Can you post the sources you are testing with? I mean the original ones, i.e. these that do not use explicitly the XA interfaces and that work correctly with the other application servers?

Best regards,

Stoyan

Message was edited by: Stoyan Vellev

[Former Member]

Hi Stoyan,

Basically the code is distributed in a number of classes, however I stripped away some code to show the idea (please let me know if this is enough).

Generally the scenario is the following:

A junit test sends two messages to a sendTestQueue through a call to a sessionbean method (transaction required). One call purposefully aborts the transaction inside the session bean method, which should prevent message to be sent. The other one does not.

A message driven bean listener takes any message that comes from sendTestQueue and sends it back to another queue (sendBackTestQueue). Then in the junit test we recieve messages from the sendBackTestQueue. Assertion is made that we must only recieve the successful message (not the failed message).

Here is the test code:

public void testSendMessageInFailedTransaction() throws Exception {
		String fail = "This message should not be received as a reply.";
		String success = "This message should be the only messaged returned in a reply.";
        String successCorrleationId = "MDBTest.testSendMessageInTransaction.SUCCESS_MSG";
        String failCorrleationId = "MDBTest.testSendMessageInTransaction.FAIL_MSG";

		assertNull("The reply queue is supposed to be empty", queueReceiver.receiveNoWait());

		// Send a message with a forced exception.  There should not be a reply to this message.		
		try {

			sessionEJB.sendMessage(fail, failCorrleationId, true);
		} catch (Exception e) {
			handleForcedException(e);
		}

		// Send a second message without an exception.  We should receive this message in a reply.		
		sessionEJB.sendMessage(success, successCorrleationId, false);

		TextMessage replyMsg = (TextMessage) queueReceiver.receive(5000);

		if (replyMsg == null)
			fail("A reply message was not received within a 5 second timeout.");
		else
			assertEquals("The reply message ID should equal the success message ID. Reply ID=" + replyMsg.getJMSCorrelationID() + ", success ID=" + successCorrleationId, successCorrleationId, replyMsg.getJMSCorrelationID());

		assertNull("The reply queue is supposed to be empty after receiving", queueReceiver.receiveNoWait());
	}

The body of the session bean method that we use to send an receive messages

public void sendMessage( String message, String correlationId, boolean forceException )
    {                
        // code for the next 2 methods is listed below 
        Message jmsMsg = base.buildMessage( message );
        Queue replyQueue = base.getQueue( JMSMessageApplicationInterface.JMS_TEST_BACK_QUEUE );
        try
        {
            jmsMsg.setJMSCorrelationID( correlationId );
            jmsMsg.setJMSReplyTo( replyQueue );
        }
        catch( JMSException jex )
        {
            throw new EJBException( jex );
        }
        
        Queue requestQueue = base.getQueue( JMSMessageApplicationInterface.JMS_TEST_QUEUE );
        base.sendMessage( requestQueue, jmsMsg );
        if( forceException ) {
            try {
                // setRollbackOnly
                // in this case the message should not be sent
                abort(new BasicBOBeanException(10000, JMSMessageApplicationInterface.FORCED_EXCEPTION_MSG));
            } catch (BasicBOBeanException e) {
                // do nothing  it is planned
                e.printStackTrace();  
            }
            throw new EJBException( JMSMessageApplicationInterface.FORCED_EXCEPTION_MSG );
        }
    }

MDB listener method:

public void onMessage( Message message )
    {
        JMSMessageApplicationInterface messageApp = 
            (JMSMessageApplicationInterface) super.getEJB( "JMSMessageApplication" );
        try
        {
            TextMessage receivedMessage = (TextMessage) message;
            
            logMessage( "Message received with the following properties:", receivedMessage );                                                                                
Queue replyQueue = (Queue) message.getJMSReplyTo();
            if( replyQueue != null )
            {            
                // Since the message is requesting a reply, create a message producer and 
                // send a reply to the reply test queue.  Copy the message content and 
                // correlation ID to the reply message so that the original sender can correlate
                // the reply with the original message.
                InitialContext ctx = JNDIUtils.getInitialContext();
                QueueConnectionFactory queueConFact = (QueueConnectionFactory) ctx.lookup( JNDIUtils.Q_CONFAC );
                QueueConnection queueCon = queueConFact.createQueueConnection();
                QueueSession queueSession = queueCon.createQueueSession( false, Session.AUTO_ACKNOWLEDGE );                
                QueueSender replySender = queueSession.createSender( replyQueue );
                queueCon.start();                        
    
                TextMessage replyMessage = queueSession.createTextMessage( receivedMessage.getText() );                
                replyMessage.setJMSCorrelationID( receivedMessage.getJMSCorrelationID() );
                replySender.send( replyMessage );
                                                                                logMessage( "Message sent with the following properties:", replyMessage );
            }                
        }
        catch( JMSException e )
        {                
            _logger.error( e );
            throw new javax.ejb.EJBException( e );
        }
        catch( NamingException e )
        {
            _logger.error( e );
            throw new javax.ejb.EJBException( e );
        }
    }

Base methods to obtain queues and build messages

  public void sendMessage( Queue queue, Message msg )
    {
        try
        {
            if ( Utils.isTraceEnabled( Utils.JMS ) )
            {
                String destName = queue.getQueueName();
                doMessageTrace( destName, msg );
            }
            QueueSender sender = getQueueSender();
            sender.send( queue, msg );
        }
        catch ( JMSException je )
        {
            throw new EJBException( je );
        }
        catch ( NamingException ne )
        {
            throw new EJBException( ne );
        }
    }

    private QueueSender getQueueSender() throws JMSException, NamingException
    {
        // open up JMS channel if not already done
        if ( _sender == null )
        {
            QueueSession session = getQueueSession();
            _sender = session.createSender( null );
        }
        return _sender;
    }


   private QueueSession getQueueSession() throws JMSException, NamingException
    {
        if ( _queueSession == null )
        {
            Context ctx = getInitialContext();
            QueueConnectionFactory factory = (QueueConnectionFactory) ctx.lookup( JNDIUtils.Q_CONFAC );
            _queueConnection = factory.createQueueConnection();
            _queueSession  = _queueConnection.createQueueSession( false, Session.AUTO_ACKNOWLEDGE );
        }
        return _queueSession;
    }

Regards,

Mikhail