[CAMEL-14405] Problem when recovering RabbitMQ connections when using a connectionFactory and the automaticRecoveryEnabled option

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Information Provided
• Affects Version/s: 2.24.3
• Fix Version/s: 3.x
• Component/s: camel-rabbitmq
• Estimated Complexity: Unknown
• Assignee: Unassigned
• Reporter: Arnaud Chotard
• Created: 15/Jan/20 11:19
• Updated: 22/Mar/21 18:02
• Resolved: 22/Mar/21 18:02

Description

The Camel RabbitMQ endpoint has a connectionFactory property allowing you to use an existing RabbitMQ connection factory.

from("rabbitmq://server:5672?connectionFactory=#brokerConnectionFactory&...");

ConnectionFactory brokerConnectionFactory = new com.rabbitmq.client.ConnectionFactory();
  brokerConnectionFactory.setHost(hostname);
  brokerConnectionFactory.setPort(port);
  brokerConnectionFactory.setVirtualHost(virtualHost);
  brokerConnectionFactory.setUsername(username);
  brokerConnectionFactory.setPassword(password);
  brokerConnectionFactory.setAutomaticRecoveryEnabled(true);

The documentation states "When this option is set, all connection options (connectionTimeout, requestedChannelMax…) set on URI are not used".

We observed problems during the recovery of connections to the broker when using the automaticRecoveryEnabled option which is present at the connectionFactory level and at the camelEndpoint level :

from("rabbitmq://server:5672?connectionFactory=#brokerConnectionFactory&automaticRecoveryEnabled=...");

1/ If camelEndpoint.automaticRecoveryEnabled = false and connectionFactory.automaticRecoveryEnabled = false => The connection and the channel are manually recreated by Camel RabbitMQ. On the other hand the doStart () is not invoked by the start () and the consumption of messages via channel.basicConsume is never reset. Messages are no longer consumed.

2/ If camelEndpoint.automaticRecoveryEnabled = true and connectionFactory.automaticRecoveryEnabled = false => Endless attempt to reconnect: Unable to obtain a RabbitMQ channel. Will try again. Caused by: Waiting for channel to re-open. Camel waits for RabbitMQ client to reset connection indefinitely. Messages are no longer consumed.

3/ If camelEndpoint.automaticRecoveryEnabled = false and connectionFactory.automaticRecoveryEnabled = true => Conflict in reopening channels by Camel Endpoint and by RabbitMQ client. Unknown delivery tag when using basickAck in doHandleDelivery and ShutdownSignalException at the initiative of the application. Messages are no longer consumed.

4/ If camelEndpoint.automaticRecoveryEnabled = true and connectionFactory.automaticRecoveryEnabled = true => It works.

The cause of 1 seems to come from the doStart() method which is not called if the consumer is already started :

@Override 
protected void doStart() throws Exception {
   if (channel == null) { 
      throw new IOException("The RabbitMQ channel is not open"); 
   }
   tag = channel.basicConsume(consumer.getEndpoint().getQueue(), consumer.getEndpoint().isAutoAck(), "", false, consumer.getEndpoint().isExclusiveConsumer(), null, this); 
}

The cause of 2 and 3 seems to come from the isAutomaticRecoveryEnabled() methods of RabbitConsumer and RabbitMQConsumer classes which tests the value of the property automaticRecoveryEnabled only on the camelEndpoint and not on the connectionFactory :

private boolean isAutomaticRecoveryEnabled() {}}
    return this.endpoint.getAutomaticRecoveryEnabled() != null
        && this.endpoint.getAutomaticRecoveryEnabled();
}

Issue Links

• relates to: CAMEL-15874 Every endpoint using RabbitMQ creates a new connection instead of using only one connection