Ewww... I hate spring xml DSL (fluent Java or Scala is much better) ..
anyway after my eyes are done bleeding ... it seems like you are wrapping
the "possible failure" calls in a doTry/catch. Since you are handling the
transactions, camel assumes all is well and that you meant to do that, so
an exception wont be propagated and the transaction wont be rolled back. If
you want to have a transaction be rolled back you have to let the exception
propagate while you have a transaction error handler in scope or
setRollbackOnly() on the transaction when you catch the exception: Such as:
from(endpointAMQ(QUEUE_DB_MONITOR_EVENT)) // Monitor the event queue
.routeId(ROUTE_ID_AUTOMATION_CASES_DB_PROCESSOR) // give the route
an ID
.onException(Exception.class).useOriginalMessage().handled(true).maximumRedeliveries(0)
// catch exceptions
.setHeader(Exchange.FAILURE_ROUTE_ID, property(Exchange.
FAILURE_ROUTE_ID)) // set route that errored
.setHeader(Exchange.EXCEPTION_CAUGHT, simple(
"${exception.stacktrace}")) // Store reason for error
.to(ExchangePattern.InOnly,
endpointAMQ(QUEUE_CASE_AUTOMATION_DEAD)).end()
// to DLQ
.transacted(KEY_TXPOLICY_REQUIRED) // make the route transacted
// rest of route
As you can see I store the exception stack trace and other information in
the headers of the failed exchange, send the exchange to the dead letter
queue and then so on. So if there is an exception in this route, NOTHING
will get rolled back which is what I want since I cant reprocess the
message. So why transacted? If someone pulls the plug on the machine or
uses kill -9 on the process, I want AMQ to retain the message. Now if I
wanted a rollback I could add the call .rollback() right after my onException()
call and before the .end call. That would tell the exception error handler
(which is a transaction handler) to rollback AMQ.
Now if you want to rollback both a database and AMQ transaction you will
need a JTA transaction manager.
*Robert Simmons Jr. MSc. - Lead Java Architect @ EA*
*Author of: Hardcore Java (2003) and Maintainable Java (2012)*
*LinkedIn: **http://www.linkedin.com/pub/robert-simmons/40/852/a39
<http://www.linkedin.com/pub/robert-simmons/40/852/a39>*
On Mon, Dec 23, 2013 at 2:11 AM, Tarun Kumar <agrawal.taru...@gmail.com>wrote:
> Hi Kraythe,
>
> Thanks for replying.
>
> Here is how my route looks like:
>
> <camel:route id="createMigrationRoute">
>
> <camel:from uri="timer://foo?period=1000"></camel:from>
>
> <camel:transacted>
>
> <camel:doTry>
>
> <camel:bean ref="abc" method="fetchMessage"></camel:bean>
>
> <camel:bean ref="pqr" method="processMessage"></camel:bean>
>
> <camel:doCatch>
>
> <camel:exception>java.lang.IOException</camel:exception>
>
> <camel:bean ref="def" method="handleException" />
>
> </camel:doCatch>
>
> <camel:doFinally>
>
> <camel:bean ref="dsd" method="me2"></camel:bean>
>
> </camel:doFinally>
>
> </camel:doTry>
>
> </camel:transacted>
>
> </camel:route>
>
>
> And beans looks like this:
>
>
> <bean id="catalogJms" class="org.apache.camel.component.jms.JmsComponent">
>
> <property name="connectionFactory">
>
> <bean class="org.apache.activemq.ActiveMQConnectionFactory">
>
> <property name="brokerURL">
>
> <value>tcp://localhost:61616</value>
>
> </property>
>
> </bean>
>
> </property>
>
> <property name="transacted" value="true"></property>
>
> <property name="transactionManager" ref="jmsTransactionManager" />
>
> <property name="acknowledgementModeName">
>
> <value>CLIENT_ACKNOWLEDGE</value>
>
> </property>
>
> </bean>
>
>
> <bean id="poolConnectionFactory" class=
> "org.apache.activemq.pool.PooledConnectionFactory">
>
> <property name="maxConnections" value="8" />
>
> <property name="connectionFactory" ref="jmsConnectionFactory" />
>
> </bean>
>
>
> <bean id="jmsConnectionFactory" class=
> "org.apache.activemq.ActiveMQConnectionFactory">
>
> <property name="brokerURL"
>
> value=
> "vm://localhost:61616?broker.persistent=false&broker.useJmx=false" />
>
> </bean>
>
>
> <!-- setup spring jms TX manager -->
>
> <bean id="jmsTransactionManager"
>
> class="org.springframework.jms.connection.JmsTransactionManager">
>
> <property name="connectionFactory" ref="poolConnectionFactory" />
>
> </bean>
>
>
> And fetchMessage in abc bean class looks like:
>
>
> public String fetchMessage() {
>
> Exchange exchange = consumer
>
> .receive(
>
> "catalogJms:queue:queue1?mapJmsMessage=false&acknowledgementModeName=CLIENT_ACKNOWLEDGE&transacted=true",
> 5000);
>
> return exchange.getIn().getBody(TextMessage.class).toString();
>
> }
>
>
> It still consumes message (QueueSize decreases by one) even when it fails
> in pqr bean with any exception apart from IOException.
>
> Please let me know what am i missing here?
>
>
>
>
> On Sat, Dec 21, 2013 at 10:01 PM, kraythe . <kray...@gmail.com> wrote:
>
> > In that case you are going about it wrong. Camel has this support built
> in
> > for you. If the route is declared to be transacted with
> > .transacted(myRequirdPolicy) then camel will handle that all for you with
> > that one line. Put the transacted line right after your exception
> handleing
> > and before the main route and then go about your business. If you want a
> > composite transaction that rolls back both the DB and JMS work then you
> > will need to configure a jta transaction manager. Just google 'camel
> > jta transacted route' and you will get code examples and so on.
> >
> > On Saturday, December 21, 2013, Tarun Kumar wrote:
> >
> > > Thanks for replying. Here is my usecase:
> > >
> > > consume a message from the queue. Do some transformation on the messge.
> > > Then, persist the transformed message to datastore.
> > > Once, transformed message is written to datastore, send ack. Reason i
> > want
> > > to send ack later is, in case my application goes down post message
> > > consumption and before writing to datastore, i should be able to fetch
> > same
> > > message again from JMS queue. Hope that clarifies.
> > >
> > > Thanks!
> > >
> > >
> > >
> > > On Sat, Dec 21, 2013 at 9:14 PM, kraythe . <kray...@gmail.com
> > <javascript:;>>
> > > wrote:
> > >
> > > > I don't think the question is quite clear. In JMS you can only
> consume
> > a
> > > > message off a queue once. You can't consume it and leave it on the
> > queue.
> > > > Why do want to not ack the message? Queues are like throwing candy
> > into a
> > > > room of kindergarten children. All will scramble for the candy but
> each
> > > > piece will get consumed by only one child. The only time JMS messages
> > > > aren't caked is when they are failed in delivery. And even then, the
> > only
> > > > use case for leaving them on the queue is because of a failed
> > transaction
> > > > in a transactional route,
> > > >
> > > > Tell me what is the use case you are trying to implement and I might
> be
> > > > able to help you down another path.
> > > >
> > > > On Saturday, December 21, 2013, Tarun Kumar wrote:
> > > >
> > > > > Hi,
> > > > >
> > > > > Any help here will be highly appreciated.
> > > > >
> > > > >
> > > > > On Fri, Dec 20, 2013 at 9:55 AM, Tarun Kumar <
> > > agrawal.taru...@gmail.com <javascript:;>
> > > > <javascript:;>
> > > > > >wrote:
> > > > >
> > > > > > I am using polling consumer.
> > > > > >
> > > > > > from("timer://foo?period=5000").bean(cool, "someBusinessLogic");
> > > > > >
> > > > > > public static class MyCoolBean {
> > > > > >
> > > > > > private ConsumerTemplate consumer;
> > > > > >
> > > > > > public void setConsumer(ConsumerTemplate consumer) {
> > > > > > this.consumer = consumer;
> > > > > > }
> > > > > >
> > > > > > public void someBusinessLogic() {
> > > > > >
> > > > > > Exchange exchange
> > > > > > =
> > > > >
> > > >
> > >
> >
> consumer.receive("catalogJms:queue:queueName?mapJmsMessage=false&acknowledgementModeName=CLIENT_ACKNOWLEDGE");
> > > > > >
> > > > > > }
> > > > > >
> > > > > >
> > > > > > Even though i am not sending acknowledgement for received
> message,
> > > each
> > > > > > time it is polling different message. Any idea why that's
> > happening?
> > > > > >
> > > > >
> > > >
> > > >
> > > > --
> > > > *Robert Simmons Jr. MSc. - Lead Java Architect @ EA*
> > > > *Author of: Hardcore Java (2003) and Maintainable Java (2012)*
> > > > *LinkedIn: **http://www.linkedin.com/pub/robert-simmons/40/852/a39
> > > > <http://www.linkedin.com/pub/robert-simmons/40/852/a39>*
> > > >
> > >
> >
> >
> > --
> > *Robert Simmons Jr. MSc. - Lead Java Architect @ EA*
> > *Author of: Hardcore Java (2003) and Maintainable Java (2012)*
> > *LinkedIn: **http://www.linkedin.com/pub/robert-simmons/40/852/a39
> > <http://www.linkedin.com/pub/robert-simmons/40/852/a39>*
> >
>
