L.S., It may be a silly question, but did you do an osgi:update <xx> for the activemq.xml bundle to ensure that the changes you made to the configuration are getting picked up? An other way to ensure picking up the new configuration is by deleting the data folder and then restarting the container.
Regards, Gert Vanthienen ------------------------ FuseSource Web: http://fusesource.com Blog: http://gertvanthienen.blogspot.com/ On Wed, Oct 26, 2011 at 6:35 PM, MarkS1900 <[email protected]>wrote: > So I have discovered some issues with my code. The following code seems to > be a working example with Atomikos (resource-context.xml) with Test Case > (ServiceMixXaRollbackTest) using a ServiceMix 4.3 ActiveMq instance > (activemq-broker.xml). > > > Unfortunately, on transaction rollback (and retry exhaustion) I am > expecting > my ActiveMq message to move to the Dead Letter Queue > "DLQ.my_test_thirdparty" as specified in the activemq-broker.xml, since I > don't override any message redelivery and dlq handling behaviour in Camel. > Instead I am finding that the ActiveMq message lands in the queue > "ActiveMQ.DLQ". > > What am I missing in my understanding on the behaviour of ActiveMq and > Camel > ? > > > Updated code below. > > > +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ > > Start code snippet: > "src/main/resources/META-INF/spring/resource-context.xml" > > +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ > > > > <?xml version="1.0" encoding="UTF-8"?> > <beans xmlns="http://www.springframework.org/schema/beans"; > xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; > xmlns:osgi="http://www.springframework.org/schema/osgi"; > xmlns:cxfse="http://servicemix.apache.org/cxfse/1.0"; > xmlns:broker="http://activemq.apache.org/schema/core"; > xmlns:tx="http://www.springframework.org/schema/tx"; > xsi:schemaLocation="http://www.springframework.org/schema/beans > http://www.springframework.org/schema/beans/spring-beans-3.0.xsd > http://camel.apache.org/schema/spring > http://camel.apache.org/schema/spring/camel-spring.xsd > http://www.springframework.org/schema/osgi > http://www.springframework.org/schema/osgi/spring-osgi.xsd > http://activemq.apache.org/schema/core > http://activemq.apache.org/schema/core/activemq-core.xsd > http://servicemix.apache.org/file/1.0 > http://servicemix.apache.org/file/1.0/servicemix-file.xsd > http://servicemix.apache.org/cxfse/1.0 > http://servicemix.apache.org/schema/servicemix-cxf-se-2011.01.xsd > http://activemq.apache.org/schema/core > http://activemq.apache.org/schema/core/activemq-core-3.0.xsd > http://www.springframework.org/schema/tx > http://www.springframework.org/schema/tx/spring-tx-3.0.xsd";> > > > > > > > > <bean id="activemq" > class="org.apache.activemq.camel.component.ActiveMQComponent"> > <property name="transacted" value="true"/> > <property name="transactionManager" ref="jtaTransactionManager"/> > </bean> > > <bean id="jms" class="org.apache.camel.component.jms.JmsComponent"> > <property name="transacted" value="true"/> > <property name="transactionManager" ref="jtaTransactionManager"/> > </bean> > > <bean id="sql" class="org.apache.camel.component.sql.SqlComponent"> > <property name="dataSource" ref="dataSource"/> > </bean> > > <bean id="jdbc" class="org.apache.camel.component.jdbc.JdbcComponent"> > <property name="dataSource" ref="dataSource"/> > </bean> > > > > > <bean id="atomikosTransactionManager" > class="com.atomikos.icatch.jta.UserTransactionManager" init-method="init" > destroy-method="close" > > <property name="forceShutdown" value="false" /> > </bean> > > > <bean id="atomikosUserTransaction" > class="com.atomikos.icatch.jta.UserTransactionImp" > > <property name="transactionTimeout" value="150" /> > </bean> > > > <bean id="jtaTransactionManager" > class="org.springframework.transaction.jta.JtaTransactionManager"> > <property name="transactionManager" ref="atomikosTransactionManager" > /> > <property name="userTransaction" ref="atomikosUserTransaction" /> > <property name="allowCustomIsolationLevels" value="true" /> > </bean> > > > <bean id="PROPAGATION_REQUIRED" > class="org.apache.camel.spring.spi.SpringTransactionPolicy"> > <property name="transactionManager" ref="jtaTransactionManager" /> > <property name="propagationBehaviorName" > value="PROPAGATION_REQUIRED" /> > <property name="transactionTemplate" ref="transactionTemplate" /> > </bean> > > <bean id="transactionTemplate" > class="org.springframework.transaction.support.TransactionTemplate"> > <property name="transactionManager" ref="jtaTransactionManager" /> > <property name="isolationLevelName" value="ISOLATION_SERIALIZABLE"/> > > </bean> > > > <bean id="jdbcTemplate" > class="org.springframework.jdbc.core.JdbcTemplate"> > <property name="dataSource" ref="dataSource"/> > </bean> > > <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"> > <property name="connectionFactory" ref="atomikosConnectionFactory" > /> > </bean> > > > > > <bean id="jmsXaConnectionFactory" > class="org.apache.activemq.ActiveMQXAConnectionFactory"> > <property name="brokerURL" value="tcp://localhost:61616" /> > </bean> > > <bean id="atomikosConnectionFactory" > class="com.atomikos.jms.AtomikosConnectionFactoryBean" init-method="init" > destroy-method="close"> > <property name="uniqueResourceName" value="amq1" /> > <property name="xaConnectionFactory" ref="jmsXaConnectionFactory" > /> > <property name="localTransactionMode" value ="false" /> > > </bean> > > > > > > <bean id="dataSource" class="com.atomikos.jdbc.AtomikosDataSourceBean"> > <property name="uniqueResourceName" value="ds1" /> > <property name="testQuery" value="select 1" /> > <property name="xaDataSource"> > <bean > class="com.microsoft.sqlserver.jdbc.SQLServerXADataSource"> > <property name="serverName" value="localhost" /> > <property name="portNumber" value="1433" /> > <property name="selectMethod" value="cursor" /> > <property name="databaseName" value="messaging" /> > <property name="user" value="__user__" /> > <property name="password" value="_password__" /> > </bean> > </property> > </bean> > > </beans> > > +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ > > End code snippet: > "src/main/resources/META-INF/spring/resource-context.xml" > > +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ > > > > > +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ > > Note 1 (Broke my transaction wrapping): > > +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ > > > <bean id="activemq" > class="org.apache.activemq.camel.component.ActiveMQComponent"> > <property name="transacted" value="true"/> > <property name="transactionManager" ref="jtaTransactionManager"/> > </bean> > > <bean id="jms" class="org.apache.camel.component.jms.JmsComponent"> > <property name="transacted" value="true"/> > <property name="transactionManager" ref="jtaTransactionManager"/> > </bean> > > > +++++++++++++++++++ IS NOT EQUIVALENT TO +++++++++++++++++++ > > > <bean id="jmsConfig" > class="org.apache.camel.component.jms.JmsConfiguration"> > <property name="transactionManager" ref="jtaTransactionManager"/> > <property name="connectionFactory" ref="atomikosConnectionFactory"/> > <property name="transacted" value="true"/> > </bean> > > > <bean id="activemq" > class="org.apache.activemq.camel.component.ActiveMQComponent"> > <property name="configuration" ref="jmsConfig"/> > </bean> > > <bean id="jms" class="org.apache.camel.component.jms.JmsComponent"> > <property name="configuration" ref="jmsConfig"/> > </bean> > > > > +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ > > Note 2 (Broke my transaction wrapping): > > +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ > > > > <bean id="dataSource" class="com.atomikos.jdbc.AtomikosDataSourceBean"> > <property name="uniqueResourceName" value="ds1" /> > <property name="testQuery" value="select 1" /> > <property name="xaDataSource"> > <bean > class="com.microsoft.sqlserver.jdbc.SQLServerXADataSource"> > <property name="serverName" value="localhost" /> > <property name="portNumber" value="1433" /> > <property name="selectMethod" value="cursor" /> > <property name="databaseName" value="messaging" /> > <property name="user" value="__user__" /> > <property name="password" value="_password__" /> > </bean> > </property> > </bean> > > +++++++++++++++++++ IS NOT EQUIVALENT TO +++++++++++++++++++ > > <bean id="atomikosDataSource" > class="com.atomikos.jdbc.AtomikosDataSourceBean"> > <property name="uniqueResourceName" value="ds1" /> > <property name="xaDataSource" ref="dataSource" /> > <property name="testQuery" value="select 1" /> > </bean> > > > <bean id="dataSource" > class="com.microsoft.sqlserver.jdbc.SQLServerXADataSource"> > <property name="serverName" value="localhost" /> > <property name="portNumber" value="1433" /> > <property name="selectMethod" value="cursor" /> > <property name="databaseName" value="messaging" /> > <property name="user" value="__user__" /> > <property name="password" value="_password__" /> > </bean> > > > > > > +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ > > Start code snippet: "src/test/java/xa/ServiceMixXaRollbackTest.java" > > +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ > > > > package test.xa; > > import junit.framework.Assert; > > import org.apache.camel.CamelContext; > import org.apache.camel.builder.RouteBuilder; > import org.apache.camel.language.XPath; > import org.apache.camel.spring.SpringRouteBuilder; > import org.apache.camel.test.CamelSpringTestSupport; > import org.junit.AfterClass; > import org.junit.BeforeClass; > import org.junit.Test; > import org.springframework.context.support.AbstractXmlApplicationContext; > import org.springframework.context.support.ClassPathXmlApplicationContext; > import org.springframework.jdbc.core.JdbcTemplate; > > > > public class ServiceMixXaRollbackTest extends CamelSpringTestSupport > { > protected JdbcTemplate jdbc; > > @Override > @BeforeClass > public void setUp() throws Exception { > super.setUp(); > > jdbc = context.getRegistry().lookup("jdbcTemplate", > JdbcTemplate.class); > > try > { > jdbc.execute( "drop table messaging.my_test_thirdparty" ); > } > catch ( Exception e ) > { > // ignore > } > jdbc.execute("create table messaging.my_test_thirdparty ( > thirdparty_id varchar(10), name varchar(128), created varchar(20), > status_code varchar(3) )"); > } > > @Override > @AfterClass > public void tearDown() throws Exception { > jdbc.execute("drop table messaging.my_test_thirdparty"); > } > > @Override > protected CamelContext createCamelContext() throws Exception { > CamelContext camelContext = super.createCamelContext(); > camelContext.addRoutes( createRouteBuilder() ); > return camelContext; > } > > > > @Override > protected AbstractXmlApplicationContext createApplicationContext() { > return new ClassPathXmlApplicationContext(new > String[]{"META-INF/spring/resource-context.xml"}); > } > > @Test > public void testXaRollbackAfterDb() throws Exception { > > Assert.assertEquals( 0, jdbc.queryForInt("select count(*) from > my_test_thirdparty")); > > String xml = "<?xml version=\"1.0\"?><thirdparty > id=\"123\"><name>THE TEST > THIRDPARTY</name><date>201110140815</date><code>200</code></thirdparty>"; > template.sendBody("activemq:queue:my_test_thirdparty", xml); > > Thread.sleep(15000); > > Assert.assertEquals( 0, jdbc.queryForInt("select count(*) from > my_test_thirdparty")); > > String dlq = > consumer.receiveBodyNoWait("activemq:queue:DLQ.my_test_thirdparty", > String.class); > Assert.assertNotNull("Should not lose message", dlq); > } > > @Override > protected RouteBuilder createRouteBuilder() throws Exception { > return new SpringRouteBuilder() { > > @Override > public void configure() throws Exception { > > from("activemq:queue:my_test_thirdparty") > .transacted() > .log("+++ Before Database Call +++") > .bean(ServiceMixXaRollbackTest.class, "toSql") > .to("jdbc:dataSource") > .log("+++ After Database Call +++") > .throwException(new IllegalArgumentException("Unexpected > Exception")) > ; > > > } > > }; > } > > > public static String toSql(@XPath("thirdparty/@id") int thirdpartyId, > @XPath("thirdparty/name/text()") String name, > @XPath("thirdparty/date/text()") long created, > @XPath("thirdparty/code/text()") int status_code) { > > if (thirdpartyId <= 0) { > throw new IllegalArgumentException("ThirdPartyId is invalid, was > " + thirdpartyId); > } > > StringBuilder sb = new StringBuilder(); > sb.append("INSERT INTO messaging.my_test_thirdparty (thirdparty_id, > name, created, status_code) VALUES ("); > sb.append("'").append(thirdpartyId).append("', "); > sb.append("'").append(name).append("', "); > sb.append("'").append(created).append("', "); > sb.append("'").append(status_code).append("') "); > > return sb.toString(); > } > > > } > > +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ > > End code snippet: "src/test/java/xa/ServiceMixXaRollbackTest.java" > > +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ > > > +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ > > Start code snippet: "apache-servicemix-4.3.0/etc/activemq-broker.xml" > ++ (Just removed comments from activemq-broker.xml, otherwise the same > source xml as the default ServiceMix installation) > > +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ > > <blueprint xmlns="http://www.osgi.org/xmlns/blueprint/v1.0.0"; > > xmlns:cm="http://aries.apache.org/blueprint/xmlns/blueprint-cm/v1.0.0"; > > xmlns:ext="http://aries.apache.org/blueprint/xmlns/blueprint-ext/v1.0.0"; > xmlns:amq="http://activemq.apache.org/schema/core";> > > <ext:property-placeholder /> > > <broker xmlns="http://activemq.apache.org/schema/core"; > brokerName="default" dataDirectory="${karaf.data}/activemq/default" > useShutdownHook="false"> > > <destinationPolicy> > <policyMap> > <policyEntries> > <policyEntry topic=">" producerFlowControl="true" > memoryLimit="10mb"> > <pendingSubscriberPolicy> > <vmCursor /> > </pendingSubscriberPolicy> > </policyEntry> > <policyEntry queue=">" producerFlowControl="true" > memoryLimit="10mb"> > > <deadLetterStrategy> > > > <individualDeadLetterStrategy queuePrefix="DLQ." > useQueueForQueueMessages="true" /> > </deadLetterStrategy> > </policyEntry> > </policyEntries> > </policyMap> > </destinationPolicy> > > <managementContext> > <managementContext createConnector="false"/> > </managementContext> > > <persistenceAdapter> > <kahaDB directory="${karaf.data}/activemq/default/kahadb"/> > </persistenceAdapter> > > <transportConnectors> > <transportConnector name="openwire" > uri="tcp://localhost:61616"/> > <transportConnector name="stomp" > uri="stomp://localhost:61613"/> > </transportConnectors> > > </broker> > > <bean id="activemqConnectionFactory" > class="org.apache.activemq.ActiveMQConnectionFactory"> > > <property name="brokerURL" value="tcp://localhost:61616" /> > </bean> > > <bean id="pooledConnectionFactory" > class="org.apache.activemq.pool.PooledConnectionFactory"> > <property name="maxConnections" value="8" /> > <property name="connectionFactory" ref="activemqConnectionFactory" > /> > </bean> > > <bean id="resourceManager" > class="org.apache.activemq.pool.ActiveMQResourceManager" > init-method="recoverResource"> > <property name="transactionManager" ref="transactionManager" /> > <property name="connectionFactory" ref="activemqConnectionFactory" > /> > <property name="resourceName" value="activemq.default" /> > </bean> > > <reference id="transactionManager" > interface="javax.transaction.TransactionManager" /> > > <service ref="pooledConnectionFactory" > interface="javax.jms.ConnectionFactory"> > <service-properties> > <entry key="name" value="localhost"/> > </service-properties> > </service> > > </blueprint> > > +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ > > End code snippet: "apache-servicemix-4.3.0/etc/activemq-broker.xml" > > +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ > > > -- > View this message in context: > http://servicemix.396122.n5.nabble.com/JMS-and-Database-interactions-under-the-same-transactional-context-tp4762819p4940331.html > Sent from the ServiceMix - User mailing list archive at Nabble.com. >
