Handling parse exceptions - how to disgard the message?

[andrew.hunt]

Hi
Struggling a fair bit here. Message Consumer listening to a topic. Normal processing is to read message, persist appropriate data and commit message. Doing this in a transaction against a durable topic. EMS and SQLServer backends. So far so good.
Error handling, however, is not going to well. Requirement is to log and disgard a message that fails to parse - there is nothing that can be done with it, so accept it and on. Unfortunately, while the logging is working, the message is being rolled back and re-presented.
Tried removing the transaction manager (JmsTransactionManager) - no change. Error Handlers only have access that I can see to the exception, not the session or anything else, so cannot influence the flow. Tried stepping through code and appeared I could unchange the status.rollbackOnly (back to false), however, I cannot find anywhere I could influence that.
Tried adding a MessageSelector implementation - but that stopped everything! I think I'm missing something there. Just did a log.info("Hi there") and return true, but complete silence.

My config is pretty simple
<int-jms: publish-subscribe-channel id="nrmlInbound" topic-name="${nrml.rate.topic.name}"
transaction-manager="jmsTransManager"
message-converter="nrmlMsgConverter"
durable="${nrml.rate.topic.durable}"
subscription="${nrml.rate.topic.subname}"
phase="1000" />

It is never getting to the message-converter.

While this might not happen in a production environment, I don't have control over the topic tree, so someone could add a message to a lower node that adheres to a different schema, meaning I would fail to process it.

Any ideas?

Thanks...Andrew

[Amol Nayak]

Are you handling the exception in your message converter (i am assuming that is your custom implementation) or throwing one from it?
Have you had a look at EMS manual to specify the max number of redelivery attempts option?

[Gary Russell]

Please put [ code ] ... [ /code ] tags around config (no spaces inside brackets).

It sounds like you are using a jms pub-sub channel as the entry point to your flow. This is incorrect.

Use a <jms:message-driven-channel-adapter/> instead.

Add an error-channel to it; in the error flow, the message will contain a payload of MessagingException, which has properties failedMessage and cause. You can do with this whatever you wish.

So long as the error flow does not re-throw the exception (or another), the transaction will commit when the error flow concludes.

When using a jms-backed channel mid-flow, you can't catch errors (it would be a little weird having an error-channel on a channel). In that case, you can segment the flow using a gateway downstream of the channel, or use a pair of outbound/inbound adapters instead of the channel and put an error channel on the inbound adapter.

[andrew.hunt]

Amol - no this is the vanilla spring implementation.

Gary,
I thought I understood what you had said - but maybe not...

What I cannot figure out is how to get the message-convertor (to marshall into a JAXB object) into play.

When I have the following configuration, ALL messages are failing - HOWEVER, they are only failing once - so progress in the right direction I think :-)

    <jpa:repositories base-package="myorg.efx.ods.nrml" />
    <bean id="nrmlMsgSelector" class="myorg.efx.ods.nrml.NrmlMessageFilter">
        <property name="currencyPairRegex" value="(AUD\.[A-Z]{3}|USD\.[A-Z]{3}|[A-Z]{3}\.USD)" />
    </bean>
    <bean id="nrmlMsgHandler" class="myorg.efx.ods.nrml.NrmlMessageHandler">
        <property name="repository" ref="nrmlRateRepository" />
    </bean>
    <bean id="jmsTransManager" class="org.springframework.jms.connection.JmsTransactionManager">
        <property name="connectionFactory" ref="connectionFactory" />
    </bean>
    <int-jms:message-driven-channel-adapter id="nrmlIn2"
        connection-factory="connectionFactory"
        destination-name="${nrml.rate.topic.name}"
        transaction-manager="jmsTransManager"
        channel="nrmlInbound"
        error-channel="exampleErrorChannel"
        pub-sub-domain="true"
        durable-subscription-name="fxODS"
        subscription-durable="true"
    />
    <oxm:jaxb2-marshaller id="marshaller" contextPath="myorg.wfs.schema.nrml._1" />
    <bean id="nrmlMsgConverter" class="org.springframework.jms.support.converter.MarshallingMessageConverter">
        <constructor-arg ref="marshaller" />
        <constructor-arg ref="marshaller" />
    </bean>
    <int:channel id="exampleErrorChannel" />
    <int:service-activator input-channel="exampleErrorChannel" ref="loggingErrorHandler" output-channel="nullChannel"/>
    <bean id="loggingErrorHandler" class="myorg.efx.ods.nrml.MyLoggingErrorHandler"/>
    <int:channel id="nrmlInbound" />
    <int:filter input-channel="nrmlInbound" ref="nrmlMsgConverter" output-channel="nrmlSelector"/>
    <int:channel id="nrmlSelector" />
    <int:filter input-channel="nrmlSelector" ref="nrmlMsgSelector" output-channel="filteredNrml"/>
    <int:channel id="filteredNrml" />
    <int:service-activator input-channel="filteredNrml" ref="nrmlMsgHandler" />

They are failing with

                ...</NRML>][Headers={timestamp=1326435491896, id=2cf1b7a6-7f03-42bb-8175-30c94c00366d, jms_redelivered=false, jms_messageId=ID:EMS-DEV2.2E8F4EC31058328B7AE:22}]
org.springframework.integration.MessageHandlingException: error occurred in message handler [org.springframework.integration.filter.MessageFilter@19b32e]
        at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:79)
        at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:114)
        at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:101)
        at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:61)
        at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:157)
        at org.springframework.integration.core.MessagingTemplate.doSend(MessagingTemplate.java:288)
        at org.springframework.integration.core.MessagingTemplate.send(MessagingTemplate.java:149)
        at org.springframework.integration.core.MessagingTemplate.convertAndSend(MessagingTemplate.java:189)
        at org.springframework.integration.gateway.MessagingGatewaySupport.send(MessagingGatewaySupport.java:183)
        at org.springframework.integration.jms.ChannelPublishingJmsMessageListener$GatewayDelegate.send(ChannelPublishingJmsMessageListener.java:423)
        at org.springframework.integration.jms.ChannelPublishingJmsMessageListener.onMessage(ChannelPublishingJmsMessageListener.java:277)
        at org.springframework.jms.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:535)
        at org.springframework.jms.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:495)
        at org.springframework.jms.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:467)
        at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.doReceiveAndExecute(AbstractPollingMessageListenerContainer.java:325)
        at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.receiveAndExecute(AbstractPollingMessageListenerContainer.java:243)
        at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.invokeListener(DefaultMessageListenerContainer.java:1058)
        at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.executeOngoingLoop(DefaultMessageListenerContainer.java:1050)
        at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.run(DefaultMessageListenerContainer.java:947)
        at java.lang.Thread.run(Thread.java:619)
Caused by: java.lang.IllegalArgumentException: result must not be null
        at org.springframework.util.Assert.notNull(Assert.java:112)
        at org.springframework.integration.filter.AbstractMessageProcessingSelector.accept(AbstractMessageProcessingSelector.java:60)
        at org.springframework.integration.filter.MessageFilter.handleRequestMessage(MessageFilter.java:103)
        at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:97)
        at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:73)
        ... 19 more

I think I'm on the right track, because when I bring the messageConverter back into play

    <int-jms:message-driven-channel-adapter id="nrmlIn2"
        connection-factory="connectionFactory"
        destination-name="${nrml.rate.topic.name}"
        transaction-manager="jmsTransManager"
        channel="nrmlSelector"
        error-channel="exampleErrorChannel"
        message-converter="nrmlMsgConverter"
        pub-sub-domain="true"
        durable-subscription-name="fxODS"
        subscription-durable="true"
    />

the message is being rolled back and re-presented and rolled back and re-presented...

Changing

<int:filter input-channel="nrmlInbound" ref="nrmlMsgConverter" output-channel="nrmlSelector"/>

to

<int:filter input-channel="nrmlInbound" ref="nrmlMsgConverter" method="fromMessage" output-channel="nrmlSelector"/>

fails as it wants a Message, but [default] SimpleMessageConverter presents a String to the channel though it has a GenericMessage.

Sorry for being thick, but what am I missing?

Thanks!

[oleg.zhurakousky]

Basically the idea is that the adapter/gateway that has an error-channel attribute will send the received error in a form of an ErrorMessage to that channel giving its subscriber a chance to do anything. And if the process of handling ErrorMessage completes successfully the initiating adapter thinks everything is fine (as if it was a complete success) otherwise it is just an error. Sor for example, such subscriber can receive ErrorMessage and return successfully with brand new non-ErrorMessage (if reply is expected), or simply return withut re-throwing exception.

Anyway, this is basically more details on what Gary explained earlier. Also, to better understand you might find these examples (from SpringOne 2011) helpful.
Erro Handling wich does demonstrates the process Gary and I described above: https://github.com/olegz/s12gx.2011/...egration/error

And flow segmentation which demonstrates what Gary was suggestion earlier where you can see how certain error are handles at the particular segment level: https://github.com/olegz/s12gx.2011/...n/segmentation

Hope that helps

[andrew.hunt]

Oleg
Thanks for your response and links.

I now have the error handling working, I just need to do the message conversion on the channel too. I'll have more of a play with gateways to see if that does what I need - if not I'll open a more specific thread with the new problem.
Andrew