ActiveMQ connection factory is probably transacted so when the route consumes it doesn't ack the message permanently so it isn't consumed correctly.
*Robert Simmons Jr. MSc. - Lead Java Architect @ EA* *Author of: Maintainable Java (Kindle <http://www.amazon.com/Maintainable-Java-Robert-Simmons-Jr-ebook/dp/B00AKHI69K>)(iTunes <https://itunes.apple.com/us/book/maintainable-java/id585666097?mt=11>)* *LinkedIn: **http://www.linkedin.com/pub/robert-simmons/40/852/a39 <http://www.linkedin.com/pub/robert-simmons/40/852/a39>* On Tue, Jun 17, 2014 at 10:59 AM, Jeff Bischoff < jbisch...@wdtablesystems.com> wrote: > Not using a queue, just an AMQ topic. Was working fine before I introduced > Camel, so I think I must be misconfiguring something in Camel. > > If I do <rollback/> everything works, but if I do <stop/>, then my > non-Camel subscriber never gets the message. It's almost as if my Camel > consumer is not actually creating a subscription on the topic, but is > instead stealing the message intended for my non-camel subscriber. > > Does it make sense that rollback would even work on this route? It isn't > marked <transacted/> and the AMQConnector has no transaction manager > defined! Yet rollback seems to work. > > <route id="configuration.topic-out"> > <from uri="table-jms:topic:configuration.topic"/> > <log message="Picked up message with id: ${id}, > destinationType: ${header.DestinationEntityType}" loggingLevel="INFO"/> > loggingLevel="INFO"/> > <choice> > <when> > > <simple>${header.DestinationEntityType} not in 'FOO,BAR'</simple> > <log message="Detected new > ${header.Name} configuration.topic message on ${header.JMSDestination} for > destination ${header.DestinationEntityType} with id: ${id}" > loggingLevel="INFO"/> > <to > uri="lc-jms:topic:configuration.topic?requestTimeout=0"/> > <log message="Successfully > processed message with id: ${id}, destinationType: > ${header.DestinationEntityType}" loggingLevel="INFO"/> > </when> > <otherwise> > <log message="(otherwise) ignoring > message with id: ${id}, destinationType: ${header.DestinationEntityType}" > loggingLevel="INFO"/> > <rollback markRollbackOnly="true" > /> > </otherwise> > </choice> > </route> > > > Results in: > > [topic-out] (Camel (camelContext) thread #9 - > JmsConsumer[configuration.topic]) Picked up message with id: > ID:TS-0007-jbischoff-2.local-50312-1403019963746-1:2:3:1:1, > destinationType: TABLE > > [topic-out] (Camel (camelContext) thread #9 - > JmsConsumer[configuration.topic]) (otherwise) ignoring message with id: > ID:TS-0007-jbischoff-2.local-50312-1403019963746-1:2:3:1:1, > destinationType: TABLE > > [EndpointMessageListener] (Camel (camelContext) thread #9 - > JmsConsumer[configuration.topic]) Execution of JMS message listener > failed. Caused by: [org.apache.camel.RuntimeCamelException - > org.apache.camel.RollbackExchangeException: Intended rollback. > Exchange[JmsMessage[JmsMessageID: > ID:TS-0007-jbischoff-2.local-50312-1403019963746-1:2:3:1:1]]] > > > > Best, > > Jeff > > > > On 6/17/14 10:59 AM, "kraythe ." <kray...@gmail.com> wrote: > > >A JMS topic will send a copy of each message to every topic. If not then > >the system violates basic JMS spec. If the system is a mainstream system > >like ActiveMQ, then that is not your problem. If you are trying to use a > >queue as a topic it doesn't work that way. > > > >A queue is like throwing candy into the middle of a class of elementary > >kids. Each piece of candy will get consumed by only one kid but they will > >scramble for the pieces of candy as fast as they can. However, if a kid > >eats a piece of candy they can't regurgitate it and put it back into the > >pile. > > > >In your case if a route consumes a message on the queue, then it can't put > >it back on the queue, the only way it gets back there is if the route > >fails > >and rolls back. > > > >Try adding a log line or a DLQ to the otherwise and see what happens. And > >don't rollback in the otherwise. If you are using a queue, create a route > >to sort the queue into 2 other queues based on the criteria and then > >another route to process the queue where the relevant messages end up. > >i.e. > >from(my > >queue).choice().when(predicate).to("jms:queue:camel-processed-queue).other > >wise().to("jms:queue:queue-processed-externally") > > > > > >*Robert Simmons Jr. MSc. - Lead Java Architect @ EA* > >*Author of: Maintainable Java (Kindle > >< > http://www.amazon.com/Maintainable-Java-Robert-Simmons-Jr-ebook/dp/B00AKH > >I69K>)(iTunes > ><https://itunes.apple.com/us/book/maintainable-java/id585666097?mt=11>)* > >*LinkedIn: **http://www.linkedin.com/pub/robert-simmons/40/852/a39 > ><http://www.linkedin.com/pub/robert-simmons/40/852/a39>* > > > > > >On Tue, Jun 17, 2014 at 9:27 AM, Jeff Bischoff > ><jbisch...@wdtablesystems.com > >> wrote: > > > >> Robert, thanks so much for replying. > >> > >> > Without that ack, the external system thinks it is not consumed and in > >> >good faith tries to redeliver. > >> > >> Okay, that seems like a good explanation for why I am getting external > >> redeliveries! > >> > >> I guess the root of my problem is that it seems like my Camel consumer > >>is > >> competing with my non-Camel topic subscribers. To your question: > >> > >> > So the question you have to decide to solve your problem is, "What is > >>the > >> > route I want my messages to take if they are not in the when > >>condition. > >> >Do > >> > they end up in a dead letter queue? Routed elsewhere? Logged to a > >>file? > >> >But > >> > they have to end up somewhere. > >> > >> I have non-Camel endpoints that will pick up the messages that don't > >>meet > >> the filter requirements. Ideally, I would have an "otherwise" that does > >> nothing with the message (but does send the ack that it has been > >> processed). These particular messages do not need to be routed by Camel > >>in > >> my system. > >> > >> However, it seems like Camel is competing with my non-Camel endpoints to > >> consumer topic messages. I would have thought each subscriber (Camel and > >> the other subscriber) would get their own copy of the topic message, and > >> that they would not compete with each other. It seems like the non-Camel > >> consumer can only process the message if I do a "rollback" in Camel. > >> > >> I'm building JUnit tests to try to figure out what I'm doing wrong, but > >> any further insight would be greatly appreciated. > >> > >> > >> Best, > >> > >> Jeff > >> > >> > >> > >> On 6/17/14 9:16 AM, "kraythe ." <kray...@gmail.com> wrote: > >> > >> >Good suggestions but you will need to check the JMS server config if > >>you > >> >are getting external redeliveries. Those happen when an external > >>system to > >> >the camel route performs a redelivery. > >> > > >> >As for your original route, there must always be a place for an > >>exchange > >> >to > >> >go or it will stop routing. In the case of your when without otherwise, > >> >what would be the path of the exchange that fails the when condition? > >>It > >> >is > >> >probably returned to the external system in some manner and not acked > >>as > >> >being delivered so it is sent again. Without that ack, the external > >>system > >> >thinks it is not consumed and in good faith tries to redeliver. > >> > > >> >Furthermore when you rollback, you are telling the host JMS system that > >> >you > >> >did not consume the message, and it will try again of course. > >> > > >> >So the question you have to decide to solve your problem is, "What is > >>the > >> >route I want my messages to take if they are not in the when > >>condition. Do > >> >they end up in a dead letter queue? Routed elsewhere? Logged to a file? > >> >But > >> >they have to end up somewhere. > >> > > >> >*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, Jun 16, 2014 at 5:56 PM, Minh Tran <darth.minhs...@gmail.com> > >> >wrote: > >> > > >> >> I have used topics in camel and they generally work the way I expect > >>it. > >> >> > >> >> I would start by leaving the cache levels at the default and > >>simplifying > >> >> your route further. eg take out the choice and send it direct from > >> >>topic to > >> >> queue. Then add the choice but send it directly to the queue without > >>the > >> >> recipient list. Also enable tracing on the camel context so you can > >>see > >> >> what is happening to your message each step of the way. > >> >> > >> >> On 17/06/2014, at 6:28 AM, Jeff Bischoff > >><jbisch...@wdtablesystems.com> > >> >> wrote: > >> >> > >> >> > Does nobody use Topics with Camel? They don't seem to work as > >> >>expected. > >> >> > > >> >> > JB > >> >> > > >> >> > On 5/30/14 10:48 AM, "Jeff Bischoff" <jbisch...@wdtablesystems.com > > > >> >> wrote: > >> >> > > >> >> >> Correction: > >> >> >> > >> >> >> 1) Why would *CACHE_CONSUMER* on the ActiveMQComponent cause > >>endless > >> >> >> external > >> >> >> redeliveries of a (filtered out) topic message, when with other > >>cache > >> >> >> settings (like CACHE_SESSION) this does not occur? > >> >> >> > >> >> >> > >> >> >> > >> >> >> > >> >> >> On 5/30/14 10:39 AM, "Jeff Bischoff" > >><jbisch...@wdtablesystems.com> > >> >> wrote: > >> >> >> > >> >> >>> Maybe I need to narrow my questions: > >> >> >>> > >> >> >>> 1) Why would CACHE_SESSION on the ActiveMQComponent cause endless > >> >> >>> external > >> >> >>> redeliveries of a (filtered out) topic message, when with other > >> >>cache > >> >> >>> settings this does not occur? > >> >> >>> > >> >> >>> 2) Why would <rollback/> prevent the external redeliveries, but > >> >><stop/> > >> >> >>> does not prevent them? > >> >> >>> > >> >> >>> > >> >> >>> Did I provide enough info below to answer these questions? Topics > >> >>just > >> >> >>> aren't working in Camel the way that I would expect! > >> >> >>> > >> >> >>> Thanks, > >> >> >>> > >> >> >>> Jeff Bischoff > >> >> >>> WDTS > >> >> >>> > >> >> >>> On 5/29/14 4:25 PM, "Jeff Bischoff" > >><jbisch...@wdtablesystems.com> > >> >> wrote: > >> >> >>> > >> >> >>>> Hi all, > >> >> >>>> > >> >> >>>> Please forgive me if this is a basic question. > >> >> >>>> > >> >> >>>> Working with Camel and JMS Topics has seemed very finicky for > >>me so > >> >> far. > >> >> >>>> If I make slight configuration changes, I see the same message > >> >> repeated > >> >> >>>> endlessly. Using jconsole, I now know these messages are > >>"external > >> >> >>>> redeliveries", i.e. the same message is being sent from JMS to > >> >>Camel > >> >> >>>> repeatedly without ever getting through. It happens in the case > >>of > >> >> >>>> messages that I am intentionally not processing due to some > >> >>criteria. > >> >> >>>> What I don't understand is why these external redeliveries keep > >> >> >>>> occurring, when I simply want this message to be dropped. > >> >> >>>> > >> >> >>>> For example, if I use CACHE_SESSION or lower cache level, the > >> >> following > >> >> >>>> route works just fine for me: > >> >> >>>> > >> >> >>>> > >> >> >>>> <route id="topics-out"> > >> >> >>>> <from uri="topic-jms:topic:projectx.>"/> > >> >> >>>> <log message="Picked up message with id: ${id}, > >> >> >>>> destinationType: ${header.DestinationEntityType}" > >> >> loggingLevel="INFO"/> > >> >> >>>> <choice> > >> >> >>>> <when> > >> >> >>>> <simple>${header.DestinationEntityType} not in > >> >> >>>> 'FOO,BAR'</simple> > >> >> >>>> <log message="Detected new ${header.Name} > >>topic > >> >> >>>> message on ${header.JMSDestination} with id: ${id}, > >> >>destinationType: > >> >> >>>> ${header.DestinationEntityType}" loggingLevel="INFO"/> > >> >> >>>> <recipientList ignoreInvalidEndpoints="false"> > >> >> >>>> > >> >><simple>lc-jms:${header.JMSDestination}</simple> > >> >> >>>> </recipientList> > >> >> >>>> </when> > >> >> >>>> </choice> > >> >> >>>> </route> > >> >> >>>> > >> >> >>>> > >> >> >>>> However, if I switch to CACHE_CONSUMER and I have even one > >>message > >> >> >>>> addressed to 'FOO' or 'BAR', I will see the endless external > >> >> >>>> redeliveries. Note that I am using a SingleConnectionFactory > >> >> >>>> specifically > >> >> >>>> for this route. > >> >> >>>> > >> >> >>>> <route id="topics-out"> > >> >> >>>> <from uri="topic-jms:topic:projectx.>"/> > >> >> >>>> <log message="Picked up message with id: ${id}, > >> >> >>>> destinationType: ${header.DestinationEntityType}" > >> >> loggingLevel="INFO"/> > >> >> >>>> <choice> > >> >> >>>> <when> > >> >> >>>> <simple>${header.DestinationEntityType} not in > >> >> >>>> 'FOO,BAR'</simple> > >> >> >>>> <log message="Detected new ${header.Name} > >>topic > >> >> >>>> message on ${header.JMSDestination} with id: ${id}, > >> >>destinationType: > >> >> >>>> ${header.DestinationEntityType}" loggingLevel="INFO"/> > >> >> >>>> <recipientList ignoreInvalidEndpoints="false"> > >> >> >>>> > >> >><simple>lc-jms:${header.JMSDestination}</simple> > >> >> >>>> </recipientList> > >> >> >>>> </when> > >> >> >>>> </choice> > >> >> >>>> </route> > >> >> >>>> > >> >> >>>> Now, if I add an Otherwise clause with a Rollback, everything > >>works > >> >> fine > >> >> >>>> again (albeit with a handful of log messages due to the > >>rollbacks): > >> >> >>>> > >> >> >>>> <route id="topics-out"> > >> >> >>>> <from uri="topic-jms:topic:projectx.>"/> > >> >> >>>> <log message="Picked up message with id: ${id}, > >> >> >>>> destinationType: ${header.DestinationEntityType}" > >> >> loggingLevel="INFO"/> > >> >> >>>> <choice> > >> >> >>>> <when> > >> >> >>>> <simple>${header.DestinationEntityType} not in > >> >> >>>> 'FOO,BAR'</simple> > >> >> >>>> <log message="Detected new ${header.Name} > >>topic > >> >> >>>> message on ${header.JMSDestination} with id: ${id}, > >> >>destinationType: > >> >> >>>> ${header.DestinationEntityType}" loggingLevel="INFO"/> > >> >> >>>> <recipientList ignoreInvalidEndpoints="false"> > >> >> >>>> > >> >><simple>lc-jms:${header.JMSDestination}</simple> > >> >> >>>> </recipientList> > >> >> >>>> </when> > >> >> >>>> <otherwise> > >> >> >>>> <rollback markRollbackOnly="true" /> > >> >> >>>> </otherwise> > >> >> >>>> </choice> > >> >> >>>> </route> > >> >> >>>> > >> >> >>>> But if I change that Rollback to a Stop instead, I get the > >>endless > >> >> >>>> external redeliveries again: > >> >> >>>> > >> >> >>>> <route id="topics-out"> > >> >> >>>> <from uri="topic-jms:topic:projectx.>"/> > >> >> >>>> <log message="Picked up message with id: ${id}, > >> >> >>>> destinationType: ${header.DestinationEntityType}" > >> >> loggingLevel="INFO"/> > >> >> >>>> <choice> > >> >> >>>> <when> > >> >> >>>> <simple>${header.DestinationEntityType} not in > >> >> >>>> 'FOO,BAR'</simple> > >> >> >>>> <log message="Detected new ${header.Name} > >>topic > >> >> >>>> message on ${header.JMSDestination} with id: ${id}, > >> >>destinationType: > >> >> >>>> ${header.DestinationEntityType}" loggingLevel="INFO"/> > >> >> >>>> <recipientList ignoreInvalidEndpoints="false"> > >> >> >>>> > >> >><simple>lc-jms:${header.JMSDestination}</simple> > >> >> >>>> </recipientList> > >> >> >>>> </when> > >> >> >>>> <otherwise> > >> >> >>>> <stop/> > >> >> >>>> </otherwise> > >> >> >>>> </choice> > >> >> >>>> </route> > >> >> >>>> > >> >> >>>> Why does rolling back prevent the unwanted message from being > >> >> >>>> redelivered, while using "stop" causes the redelivery to occur? > >> >> >>>> Intuitively, I would have thought it would work the opposite > >>way. > >> >> >>>> Rolling > >> >> >>>> back should put the item back on the "from" endpoint, allowing > >>it > >> >>to > >> >> be > >> >> >>>> processed again. "Stop" should just drop the message, preventing > >> >> further > >> >> >>>> redelivery. What I'm actually seeing is the opposite of this. > >>Am I > >> >> >>>> misunderstanding? > >> >> >>>> > >> >> >>>> I also get the same exact symptoms if I leave the cache level at > >> >> >>>> CACHE_SESSION, but I make the subscription durable. In that case > >> >>the > >> >> >>>> rollbacks prevent the problem, but otherwise I get the eternal > >> >> external > >> >> >>>> redeliveries. > >> >> >>>> > >> >> >>>> Using a Filter element also produces the same results as using > >>the > >> >> >>>> Choice > >> >> >>>> element above: > >> >> >>>> > >> >> >>>> <route id="topics-out"> > >> >> >>>> <from > >> >> >>>> > >> >>uri="topic-jms:topic:projectx.>?durableSubscriptionName=testdurasub"/> > >> >> >>>> <log message="Picked up message with id: ${id}, > >> >> >>>> destinationType: ${header.DestinationEntityType}" > >> >> loggingLevel="INFO"/> > >> >> >>>> <filter> > >> >> >>>> <simple>${header.DestinationEntityType} not in > >> >> >>>> 'FOO,BAR'</simple> > >> >> >>>> <recipientList ignoreInvalidEndpoints="false"> > >> >> >>>> > >><simple>lc-jms:${header.JMSDestination}</simple> > >> >> >>>> </recipientList> > >> >> >>>> </filter> > >> >> >>>> </route> > >> >> >>>> > >> >> >>>> When I look at the topic in JMS, I see the topic "projectx.>" > >> >> appearing > >> >> >>>> and disappearing in the list rapidly. I assume that's caused by > >>the > >> >> >>>> Camel > >> >> >>>> polling frequency. It seems a little disturbing to me; is that > >> >>normal? > >> >> >>>> > >> >> >>>> Thanks so much for taking the time to help! > >> >> >>>> > >> >> >>>> Jeff Bischoff > >> >> >>>> WDTS > >> >> >>> > >> >> >> > >> >> > > >> >> > ________________________________ > >> >> > > >> >> > Please take note: This email, including attachments, contains > >> >> information which may be confidential or legally privileged and is > >>only > >> >>for > >> >> the use of the individual or entity to whom it is properly addressed. > >> >>Any > >> >> unauthorized review, use, disclosure, copying, or distribution is > >> >> prohibited. If you have reason to believe that you have received this > >> >> communication in error, or that it may be misaddressed or not > >>intended > >> >>for > >> >> you, please destroy it and notify the sender immediately. Thank you. > >> >> > >> >> > >> > >> ________________________________ > >> > >> Please take note: This email, including attachments, contains > >>information > >> which may be confidential or legally privileged and is only for the use > >>of > >> the individual or entity to whom it is properly addressed. Any > >>unauthorized > >> review, use, disclosure, copying, or distribution is prohibited. If you > >> have reason to believe that you have received this communication in > >>error, > >> or that it may be misaddressed or not intended for you, please destroy > >>it > >> and notify the sender immediately. Thank you. > >> > > ________________________________ > > Please take note: This email, including attachments, contains information > which may be confidential or legally privileged and is only for the use of > the individual or entity to whom it is properly addressed. Any unauthorized > review, use, disclosure, copying, or distribution is prohibited. If you > have reason to believe that you have received this communication in error, > or that it may be misaddressed or not intended for you, please destroy it > and notify the sender immediately. Thank you. >
