Error handling and redelivery in conjunction with XA transactions

[0xaf]

Hi All!

I'm trying to implement a simple and classic integration pattern which uses XA-transactions. JBoss Fuse ESB version is 6.1.0.GA.

And I need an advice how to implement error handling and redelivery in conjunction with XA transactions according to current best practices.

Example XA-transaction looks like this:

1. Receive message from ActiveMQ queue.
2. Transform message.
3. Update first database (Oracle).
4. Transform message.
5. Update second database (DB2).
6. Send message to another ActiveMQ queue.

Need to handle errors and repeat failed transaction after delay.

Looks like there are two options to implement redelivery:

1. ActiveMQ broker redelivery. Can be configured in the ConnectionFactory with RedeliveryPolicy. Redelivery provided by broker, when message consuming transaction rolled back.

   <bean id="amqXaConnectionFactory" class="org.apache.activemq.ActiveMQXAConnectionFactory">

        <property name="brokerURL" value="vm:amq"/>

        <property name="userName" value="admin"/>

        <property name="password" value="admin"/>

        <property name="prefetchPolicy">

            <bean class="org.apache.activemq.ActiveMQPrefetchPolicy">

                <property name="all" value="1"/>

            </bean>

        </property>

        <property name="redeliveryPolicy">

            <bean class="org.apache.activemq.RedeliveryPolicy">

                <property name="maximumRedeliveries" value="3"/>

                <property name="initialRedeliveryDelay" value="1000"/>

                <property name="redeliveryDelay" value="1000"/>

            </bean>

        </property>

    </bean>

2. Camel ErrorHandler redelivery. Can be configured for any camel route, for example using Java DSL. Redelivery implemented inside Camel pipeline.

       errorHandler(

                transactionErrorHandler()

                        .maximumRedeliveries(3)

                        .redeliveryDelay(1000L)

        );

But there are some architectural issues in Fuse ESB which makes both of the available redelivery methods unusable in conjunction with XA transactions:

1. Because broker redelivery has affinity to MessageConsumer, ActiveMQComponent.cacheLevelName should be set to CACHE_CONSUMER for normal operation, but it is not compatible with XA transactions which requires cache level CACHE_CONNECTION or CACHE_NONE. So RedeliveryPolicy is not provided as it should: redeliveryDelay, exponentialBackOff are not working.
2. Camel provides redelivery only from the point of failure in the Camel route pipeline. Hence active XA transaction is not rolled back on each retry loop and remains the same. Not every Camel route step can be idempotent, so this can lead to duplicate operations in XA transaction scope. For example, if failure occurs in step (3) of the route somwhere in the code after already completed SQL insert, on next redelivery this insert will be executed one more time and there will be duplicates in DB.

I spent a few days in the research and decided that only way to achieve desired classic behavior is to disable any redeliveries and implement custom Camel route for error handling. This route will read messages from ActiveMQ.DLQ queue where PERSISTENT messages are moved after XA transaction is rolled back. And then resubmit they with desired AMQ_SCHEDULED_DELAY header. But now I'm stuck on how to determine OriginalDestination from message in ActiveMQ.DLQ. When message is read with Camel route, OriginalDestination header somehow is empty, test it with:

        from("activemq-xa:queue:ActiveMQ.DLQ").routeId("catch-errors")

                .log("##### OriginalDestination = '${header.OriginalDestination}'")

I'll be grateful for advices!

[sbrandt]

Hi Andy,

what you describe in 1) is consumer redelivery policy and you're right - it only works with CACHE_CONSUMER and therefore is incompatible with XA.

Broker redelivery on the other hand works on broker itself, independent of consumers and might be what you were looking for. It needs to be configured in the broker's conf/activemq.xml, see last chapter "Broker Redelivery (v5.7)" here: [1]

[1] http://activemq.apache.org/message-redelivery-and-dlq-handling.html

[0xaf]

Hi, Sebastian!

Yes, I try this before with no luck, but seems like I missed impotant thing due to disambiguation in the documentation:

https://access.redhat.com/documentation/en-US/Red_Hat_JBoss_A-MQ/6.0/html-single/Client_Connectivity_Guide/

Configuring the broker's redelivery plug-in

Configuring a broker's redelivery plug-in is a good way to tune the redelivery of messages to all of the consumer's that use the broker. The broker sends its configuration to clients when they connect and the client will use the broker's configuration by default.

I realized that configuring redelivery client ConnectionFactory and Redelivery Plug-In is the same thing and uses same redelivery way.

Seems like that it is not the same.

In newer version of AMQ documentation there is important point:

https://access.redhat.com/documentation/en-US/Red_Hat_JBoss_A-MQ/6.1/html-single/Client_Connectivity_Guide/

Configuring the broker's redelivery plug-in

Configuring a broker's redelivery plug-in is a good way to tune the redelivery of messages to all of the consumer's that use the broker. When using the broker's redelivery plug-in, it is recommended that you disable redelivery on the consumer side (if necessary, by setting maximumRedeliveries to 0 on the destination).

So I did the following things, and got the desired behavior with working redeliveryDelay!

1. Configure ActiveMQ broker Redelivery Plug-In in etc/activemq.xml. Do not forget to enable Scheduler with schedulerSupport attribute!

   <broker ... start="true" restartAllowed="false" schedulerSupport="true" persistent="true">

                 ....

        <plugins>

            <jaasAuthenticationPlugin configuration="karaf"/>

            <redeliveryPlugin fallbackToDeadLetter="true" sendToDlqIfMaxRetriesExceeded="true">

                <redeliveryPolicyMap>

                    <redeliveryPolicyMap>

                        <redeliveryPolicyEntries/>

                        <defaultEntry>

                            <redeliveryPolicy maximumRedeliveries="3" initialRedeliveryDelay="5000" redeliveryDelay="5000"/>

                        </defaultEntry>

                    </redeliveryPolicyMap>

                </redeliveryPolicyMap>

            </redeliveryPlugin>

        </plugins>

2. Disable any redeliveries at the Camel route ErrorHandler level.

       errorHandler(

                transactionErrorHandler()

                        .disableRedelivery()

        );

3. Disable any redeliveries at ActiveMQ client level in the ConnectionFactory.

   <bean id="amqXaConnectionFactory" class="org.apache.activemq.ActiveMQXAConnectionFactory">

        <property name="brokerURL" value="vm:amq"/>

        <property name="userName" value="admin"/>

        <property name="password" value="admin"/>

        <property name="prefetchPolicy">

            <bean class="org.apache.activemq.ActiveMQPrefetchPolicy">

                <property name="all" value="1"/>

            </bean>

        </property>

        <property name="redeliveryPolicy">

            <bean class="org.apache.activemq.RedeliveryPolicy">

                <property name="maximumRedeliveries" value="0"/>

            </bean>

        </property>

    </bean>

Thanks for help!

[sbrandt]

Hi Andy, thanks sharing your configuration! You might skip disableRedelivery() on the transactionErrorHandler, I think it doesn't do redeliveries by default.