[CAMEL-11530] RabbitMQ component doesn't handle network failure properly when using autoAck=false and automaticRecovery=true

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Implemented
• Affects Version/s: 2.17.6
• Fix Version/s: 2.21.0
• Component/s: camel-rabbitmq

Environment

RabbitMQ 3.5.7

• Estimated Complexity: Unknown
• Regression: Regression
• Assignee: Claus Ibsen
• Reporter: Arnaud CHOTARD
• Created: 11/Jul/17 07:38
• Updated: 02/Feb/18 09:02
• Resolved: 02/Feb/18 09:02

Description

When using autoAck=false and automaticRecoveryEnabled=true on Camel RabbitMQ consumer, we are facing problem during network failure recovery on our production environments.

Before CAMEL-8742, tested with Camel 2.16.2, the automatic recovery is managed directly by RabbitMQ native client. When network failure is recovering, the old channel is reconnect and messages are proccessed correctly.

After CAMEL-8742, tested with Camel 2.17.6, the Camel RabbitMQ component open new channel when network failure is recovering. On the other hand, RabbitMQ native client will also reconnect old channels when using automaticRecoveryEnabled=true. So two channel are now open. The two recovery method make conflict. As consequence, new messages publish in RabbitMQ remain unacked and the Camel exchange are not processing on Camel route.

To recover from this, we should delete, recreate the queue in production environment and restart Camel application. So, the unacked messages are being lost.

What do you think of this problem ?

Thank you

Camel RabbitMQ Consumer Route :

from("rabbitmq:///?" + 
		"connectionFactory=#brokerConnectionFactory" + 
		"&addresses=" + brokerAdresses + 
		"&queue="+ brokerQueue +
		"&declare=false" + 
		"&autoDelete=false" +
		"&autoAck=false"
	)
	.bean("<beanName>", "<beanMethod>");

RabbitMQ Connection Factory with automaticRecoveryEnabled=true

    <bean id="brokerConnectionFactory" class="com.rabbitmq.client.ConnectionFactory">
        <property name="host" value="${broker.hostname}"/>
        <property name="virtualHost" value="${broker.virtualHost:/}"/>
        <property name="port" value="${broker.port:5672}"/>
        <property name="username" value="${broker.username}"/>
        <property name="password" value="${broker.password}"/>
        <property name="automaticRecoveryEnabled" value="true"/>
    </bean>