Camel JMS XA Client

classic Classic list List threaded Threaded
1 message Options
Reply | Threaded
Open this post in threaded view
|

Camel JMS XA Client

dpravin
Hi All,

Requirement - Transacted route, we have to consume messages from MQ queue. Multiple resources involved, hence use XA transactions.

Environment - Jboss Fuse 6.2.1, JDK 8, MQ 7.5.0.3

I use following to create the JMSComponent in blueprint and everything works fine as expected.


    <bean id="wmqxa" class="org.apache.camel.component.jms.JmsComponent">
        <property name="configuration" ref="WMQConfig" />
    </bean>

    <bean id="WMQConfig" class="org.apache.camel.component.jms.JmsConfiguration">
        <property name="connectionFactory" ref="FuseWmqXaConnectionFactory" />
        <property name="transacted" value="false" />
        <property name="transactionManager" ref="platformTxManager" />
        <property name="maxConcurrentConsumers" value="5" />
        <property name="cacheLevelName" value="CACHE_NONE" />
    </bean>

    <bean id="FuseWmqXaConnectionFactory" class="org.apache.activemq.jms.pool.JcaPooledConnectionFactory">
        <property name="connectionFactory" ref="WMQConnectionFactory"/>
        <property name="transactionManager" ref="recoverableTxManager"/>
        <property name="maxConnections" value="1"/>
        <property name="name" value="wmq" />
    </bean>

   
    <bean id="WMQConnectionFactory" class="com.ibm.mq.jms.MQXAConnectionFactory">
        <property name="hostName" value="hostname" />
        <property name="port" value="1414" />
        <property name="queueManager" value="QUEUE.MANAGER" />
        <property name="channel" value="SYSTEM.ADMIN.SVRCONN" />
        <property name="transportType" value="1" />
    </bean>


With respect to above configuration I created a custom bean that extends from JMSComponent and tried to create JMS XA connection factory

1) defined a custom bean com.ai.jms.WmqJmsComponent that extends from JMSComponent
2) Inside this bean I am instantiating and wiring all above components

The XA connection works fine when I send the message to MQ, however when I am using the same connection to consume message from MQ it produces following warnings/errors in log file,


PooledSession                    | 189 - org.apache.activemq.activemq-osgi - 5.11.0.redhat-621107 | Caught exception trying close() when putting session back into the pool, will invalidate. com.ibm.msg.client.jms.DetailedIllegalStateException: JMSCC0020: This session is closed.
n  | An application called a method that must not be used after the session is closed.
n  | Ensure that the session is not closed before calling the method.
com.ibm.msg.client.jms.DetailedIllegalStateException: JMSCC0020: This session is closed.
An application called a method that must not be used after the session is closed.
Ensure that the session is not closed before calling the method.
        at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)[:1.8.0_25]
        at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)[:1.8.0_25]
        at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)[:1.8.0_25]
        at java.lang.reflect.Constructor.newInstance(Constructor.java:408)[:1.8.0_25]
        at com.ibm.msg.client.commonservices.j2se.NLSServices.createException(NLSServices.java:319)[357:com.ibm.msg.client.osgi.commonservices.j2se:7.5.0.3]
        at com.ibm.msg.client.commonservices.nls.NLSServices.createException(NLSServices.java:233)[358:com.ibm.msg.client.osgi.jms:7.5.0.3]
        at com.ibm.msg.client.jms.internal.JmsErrorUtils.createException(JmsErrorUtils.java:109)[358:com.ibm.msg.client.osgi.jms:7.5.0.3]
        at com.ibm.msg.client.jms.internal.State.checkNotClosed(State.java:150)[358:com.ibm.msg.client.osgi.jms:7.5.0.3]
        at com.ibm.msg.client.jms.internal.JmsSessionImpl.checkNotClosed(JmsSessionImpl.java:2526)[358:com.ibm.msg.client.osgi.jms:7.5.0.3]
        at com.ibm.msg.client.jms.internal.JmsSessionImpl.setMessageListener(JmsSessionImpl.java:1730)[358:com.ibm.msg.client.osgi.jms:7.5.0.3]
        at com.ibm.mq.jms.MQSession.setMessageListener(MQSession.java:1004)[360:com.ibm.msg.client.osgi.wmq:7.5.0.3]
        at org.apache.activemq.jms.pool.PooledSession.close(PooledSession.java:103)[189:org.apache.activemq.activemq-osgi:5.11.0.redhat-621107]
        at org.springframework.jms.support.JmsUtils.closeSession(JmsUtils.java:108)[215:org.apache.servicemix.bundles.spring-jms:3.2.12.RELEASE_2]
        at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.clearResources(DefaultMessageListenerContainer.java:1142)[215:org.apache.servicemix.bundles.spring-jms:3.2.12.RELEASE_2]
        at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.run(DefaultMessageListenerContainer.java:1001)[215:org.apache.servicemix.bundles.spring-jms:3.2.12.RELEASE_2]
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)[:1.8.0_25]
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)[:1.8.0_25]
        at java.lang.Thread.run(Thread.java:745)[:1.8.0_25]


Appreciate any help to resolve this issue.

Thanks,
Pravin
- Pravin