Hi, I just want to ask if your RAW message is stream based. From your description I double the raw message was just consumed when it was stored into DB.
-- Willem Jiang Red Hat, Inc. Web: http://www.redhat.com Blog: http://willemjiang.blogspot.com (http://willemjiang.blogspot.com/) (English) http://jnn.iteye.com (http://jnn.javaeye.com/) (Chinese) Twitter: willemjiang Weibo: 姜宁willem On Sunday, September 15, 2013 at 2:30 AM, bhaumikananda wrote: > I am using Camel 2.10.3 in my current project. Use case - I have a program > which listens to income messages then persists the raw trade message into a > DB and split the message into small sections based on some expression. This > small sections are then transformed based on business logic and send to > downstream. > > Problem - There are cases observed where after the message being persisted > nothing else is performed, neither no trace of exception. On detecting the > trade I replayed the raw message through the queue, and it was processed > successfully. > > In my program I heavily depend on JmsTimeStamp of the exchange header. The > route is mentioned below. > Any suggestion would be helpful. > > > Route - > > <route id="fitOptInMsgRoute"> > <routeContext id="fitOptsRoutesCtx" > xmlns="http://camel.apache.org/schema/spring";> > <from ref="fitOptInMsgQueue"/> > <transacted ref="PROPAGATION_REQUIRED_FIT_OPTION"/> > <setProperty propertyName="jmsxDeliveryCount"> > <constant>{{mySystem.route.jmsx.delivery.count}}</constant> > </setProperty> > <choice> > <when> > <simple>${in.header.JMSXDeliveryCount} != null and > ${property.jmsxDeliveryCount} > ${in.header.JMSXDeliveryCount} </simple> > <process ref="fitOptStpIdValidator"/> > <process ref="fitOptExchangeHelper"/> > <doTry> > <bean ref="rawMsgSqlInputBuilder" method="populateFITOptParam"/> > <to uri="{{raw.message.insertSql.uri}}"/> > <doCatch> > <exception>org.springframework.dao.DuplicateKeyException</exception> > <log message="Duplicate Raw trade message..." loggingLevel="INFO" > logName="com.bbb.sbrs.mySystem.util.FITOptInLogger"/> > </doCatch> > </doTry> > <process ref="revertExchangeProcessor"/> > <log message="Raw FIT Trade Message persisted and processing..." > loggingLevel="INFO" logName="com.bbb.sbrs.mySystem.util.FITOptInLogger"/> > <split parallelProcessing="true" stopOnException="false" > strategyRef="fitOptAggregator"> > <xpath>/insertDeal/request/dealData/trade/trade</xpath> > <log message="Processing message after split" loggingLevel="INFO" > logName="com.bbb.sbrs.mySystem.util.FITOptInLogger" /> > <setProperty propertyName="SID"> > <simple>SOME_SID</simple> > </setProperty> > <to uri ="direct:sendFITOptTrdToSBR"/> > </split> > </when> > <otherwise> > <log message="This message has been exceed number of retry. Exhausting > it now with body.${body}." > loggingLevel="ERROR" > logName="com.bbb.sbrs.mySystem.util.FITOptInLogger" /> > </otherwise> > </choice> > </route> > > <route id="sendFITOptTrdToSBR"> > <from uri="direct:sendFITOptTrdToSBR" /> > <onException useOriginalMessage="true"> > <exception>java.lang.Exception</exception> > <handled><constant>true</constant></handled> > <process ref="aggregateExcepHelper" /> > <bean ref="excepLogInputPopulator" method="populateFITOptExcepParam"/> > <to uri="direct:exceptionHandler" /> > </onException> > <process ref="fitOptTradeMsgFilter"/> > <process ref="fitOptMessageValidator"/> > <process ref="tradeMsgInfoPopulator"/> > > <to uri="direct:trdMsgPersistence"/> > <process ref="fitOptXsltProcHelper"/> > <choice> > <when> > <xpath>@somExpression</xpath> > <log message="Applying transformation logic for xyz ." > loggingLevel="INFO" > logName="com.bbb.www.sbrs.mySystem.util.InLogger > (http://www.sbrs.mySystem.util.InLogger)"/> > <to uri="xslt:xsl/mapping1.xsl" /> > </when> > <when> > <xpath>@someExpression</xpath> > <log message="Applying transformation logic for abc." > loggingLevel="INFO" > logName="com.bbb.www.sbrs.mySystem.util.InLogger > (http://www.sbrs.mySystem.util.InLogger)"/> > <to uri="xslt:xsl/mapping2.xsl" /> > </when> > <when> > <xpath>@someExpression</xpath> > <log message="Applying transformation logic for pqr." > loggingLevel="INFO" > logName="com.bbb.www.sbrs.mySystem.util.InLogger > (http://www.sbrs.mySystem.util.InLogger)"/> > <to uri="xslt:xsl/mapping3.xsl" /> > </when> > <when> > <xpath>@someExpression</xpath> > <log message="Applying transformation logic for wxy." > loggingLevel="INFO" > logName="com.bbb.www.sbrs.mySystem.util.InLogger > (http://www.sbrs.mySystem.util.InLogger)"/> > <to uri="xslt:xsl/option/mapping4.xsl" /> > </when> > <when> > <xpath>@someExpression and @someExpression1</xpath> > <log message="Applying transformation logic for hedge." > loggingLevel="INFO" > logName="com.bbb.www.sbrs.mySystem.util.InLogger > (http://www.sbrs.mySystem.util.InLogger)"/> > <to uri="xslt:xsl/cash/mapping5.xsl" /> > </when> > </choice> > <to uri="xslt:xsl/util/RemoveNameSpace.xsl"/> > <log message="Message trnsformed.${body}" loggingLevel="WARN" > logName="com.bbb.sbrs.mySystem.util.FITOptInLogger" /> > <doTry> > <process ref="updateTransformedTradeSqlBuilder" /> > <to uri="{{trade.message.updateTransformedSql.uri}}" /> > <doCatch> > <exception>java.lang.Exception</exception> > <log message="Error occurred while updating transformed sbrs out > message in DB." > loggingLevel="ERROR" > logName="com.bbb.sbrs.mySystem.util.FITOptInLogger" /> > <bean ref="excepLogInputPopulator" method="populateFITOptExcepParam"/> > <to uri="direct:exceptionHandler" /> > </doCatch> > </doTry> > <process ref="idempUtilSBR"/> > <idempotentConsumer eager="false" > messageIdRepositoryRef="reportingMsgIdRepo"> > <header>messageId</header> > <process ref="sbrsOutProcessor"/> > <to ref="sbrsOutMsgQueue" /> > </idempotentConsumer> > <doTry> > <process ref="updateSBRSentStatSqlBuilder" /> > <to uri="{{trade.message.updateSBRSentStatusSql.uri}}" /> > <doCatch> > <exception>org.springframework.dao.DataAccessException</exception> > <log message="Error occurred while updating sent to sbrs status in DB." > loggingLevel="ERROR" > logName="com.bbb.sbrs.mySystem.util.FITSBROutLogger" /> > <bean ref="excepLogInputPopulator" method="populateFITOptExcepParam"/> > <to uri="direct:exceptionHandler" /> > </doCatch> > </doTry> > </route> > </routeContext> > > > > > -- > View this message in context: > http://camel.465427.n5.nabble.com/My-camel-route-not-able-to-handle-exception-tp5739381.html > Sent from the Camel - Users mailing list archive at Nabble.com > (http://Nabble.com).
