SimpleMessageListenerContainer error handler

[jackett_dad]

I've got a simple example working where a message is posted as xml to a fanout exchange, and I have a test client that receives the message. But when I try to retrieve the same message in my deployed system, there is an error that occurs while AMQP's SimpleMessageListenerContainer handles it. By tracing into the code, I can see that the xml string is not being transformed into a java object as I thought it would, and there is a classcastexception (I'm not asking about this) that results in the same message being added to the the source fanout queue repeatedly forever. I haven't let it run for too long, but I quickly get up to 40k messages.

I've tried putting in my own error handler, but I can't get an error handler in the listener object, which is an instance of org.springframework.amqp.rabbit.AbstractMessageLis tenerContainer. The value is always null.

I read the following here:

If you are using a SimpleMessageListenerContainer you will also be able to inject a Spring ErrorHandler instance that can be used to react to an exception in the listener. The ErrorHandler cannot prevent the exception from eventually propagating, but it can be used to log or alert another component that there is a problem.

So my questions are, how do I set this error handler so that I can at least see what error is causing the problem, and how can I stop this propagation?

Here is my current configuration:

    <rabbit:queue id="statusEventQueue" name="status.event.queue" durable="true"/>

    <rabbit:connection-factory id="connectionFactory" host="#{config.host}"
    		username="#{config.user}" password="#{config.password}" port="#{config.port}"/>

    <rabbit:template id="amqpTemplate" connection-factory="connectionFactory" />

    <rabbit:admin connection-factory="connectionFactory" />

    <!-- Fanout queue for status events.  A perl script sends a message to this exchange to update
         status of the initialization process -->
    <rabbit:fanout-exchange name="vtl.status.fanout.exchange" durable="true">
        <rabbit:bindings>
            <rabbit:binding queue="statusEventQueue"/>
        </rabbit:bindings>
    </rabbit:fanout-exchange>

    <bean id="eventMarshaller"
        class="org.springframework.oxm.jaxb.Jaxb2Marshaller">
        <property name="classesToBeBound">
            <list>
                <value>beans.NodeStatus</value>
            </list>
        </property>
    </bean>

    <bean id="unmarshallingTransformer"
          class="org.springframework.integration.xml.transformer.UnmarshallingTransformer">
        <constructor-arg ref="eventMarshaller"/>
    </bean>

    <bean id="errorHandler" class="handlers.ErrorHandler"/>

    <int-amqp:inbound-channel-adapter channel="statusControllerEventChannel"
                                      queue-names="status.event.queue"
                                      connection-factory="connectionFactory" error-handler="errorHandler"/>

    <int:channel id="eventNodeStatusOutputChannel"/>

    <int:transformer ref="unmarshallingTransformer" input-channel="statusControllerEventChannel"
                     output-channel="eventNodeStatusOutputChannel"/>

    <bean id="headStatusMap" class="beans.NodeStatusMap" factory-method="getInstance"/>

    <int:service-activator input-channel="eventNodeStatusOutputChannel" ref="headStatusMap"
                           method="handleStatusMessage"/>

I thought putting the error handler in the service activator might do the trick, but I'm thinking it has to be done in the initial rabbit settings--I'm not sure. I know this can be done if I programmatically set everything up following some of the examples in the spring-amqp project, and I could probably obtain the listener somehow in my java code through spring, but I'm sure there must be a way here.

Thanks for any help you can provide,

Scott

[jackett_dad]

The disparity between my test and production environment boiled down to my war not being redeployed by Tomcat. Now I can set the error handler in the inbound-channel-adapter element. Now I'm going to add the error-channel as per this link.

I'll update if this fixes the re-issuance of the message.

[jackett_dad]

It looks like by setting both an error handler and routing errors to the error channel of my choice, the error no longer propagates to the default handler that republishes the message. Here is the modified (partial) configuration:

<int:channel id="errorChannel"/>

<bean id="unmarshallingTransformer"
class="org.springframework.integration.xml.transfo rmer.UnmarshallingTransformer">
<constructor-arg ref="eventMarshaller"/>
</bean>

<!-- My own error handler, implements org.springframework.util.ErrorHandler -->
<bean id="myErrorHandler" class="handlers.ErrorHandler"/>

<int-amqp:inbound-channel-adapter channel="statusControllerEventChannel"
queue-names="status.event.queue"
connection-factory="connectionFactory"
error-handler="myErrorHandler"
error-channel="errorChannel"/>

<int:logging-channel-adapter id="logger" level="ERROR" channel="errorChannel"/>

I've been struggling with this for a while, so perhaps this will be helpful to someone else.

Scott

[Mark Fisher]

Scott, Thanks for posting that.

A somewhat related option is to add a RetryInterceptor on the listener (via the "advice chain"). Then you can configure the max retries, delay interval, and a handler for the case where the max retries is hit.

[jackett_dad]

Thanks, Mark. I think it would be a good idea to have that functionality in the application anyway. I'll look into it and post my modified configuration when I do.

[bwfrieds]

I'll look into it and post my modified configuration when I do.

Scott, did you try, with success, to implement a retry interceptor on the listener via the advice chain? Any luck with that? -Ben

[jackett_dad]

Ben,

I haven't tried that yet. Since I was able to harness errors as they occurred, this moved to the back of my priority list. I will come back to it eventually, once I've successfully adapted an existing gateway interface (implemented as another remote technology) to messaging. I'm getting close. Sorry I couldn't provide a solution at the moment. If you make it happen, post it here; otherwise, I'll do the same when I come back to the task.

[bwfrieds]

... If you make it happen, post it here;

Scott, I have abandoned this effort; I have a deployment deadline that will not budge. I just tested a sufficient work-around solution. In my MessageHandler.handleMessage method, I wrap the body with a try/catch. For any Exception caught during processing, I then use an autowired rabbit/amqp template bean to convertAndSend the message payload to a queue dedicated to bad messages, with a description of the Exception caught. So this poor-man's solution does not have a retry mechanism, which I really wanted. Later, I may try moving the error description to a header and out of the message payload. -Ben

[jackett_dad]

I'm sorry you had to bail on that solution. I've been fortunate that my deadline keeps moving back due to an ever-extending date to the previous release of our software. When I figure it out I'll post the answer. Glad you were able to work around the issue anyway.