Cluster rabbitmq node synchronization

[ridi2183]

I use a cluster RabbitMQ with two nodes (Node A and Node B). I use classic queues, exchanges and bindings. The queues are created on Node A and are visible and reachable from Node B. These queues, exchanges and bindings are created by my java client currently connected to Node A.

Everything works correctly when we send message to exchange using routing key.

The problem occurs when the Node A crashes. My java client lost connection, connect to Node B and recreate queues, exchanges and bindings on the Node B.
After that, some bindings correctly works but some others cause a routing error from exchange.

Is it normal ? Am I doing something wrong?

Could you please tell me how to fix this problem ?

[acogoluegnes]

Please use discussions for questions, not issues.

We need more information to help: broker and client versions, logs, stack trace if any, steps and code samples to reproduce.

Note you should only use odd numbers of nodes for a cluster (3, 5, 7, etc), not even numbers.

[ridi2183]

Thank you for your answer, here is more information:
Versions:
- OS: Oracle Linux 8.5
- RabbitMQ: 3.10.7
- Erlang: 24.3.4.4
- Java amqp-client: 5.15.0
Log:
- no specific logs

Cluster:
- Note that I use 2 RabbitMQ nodes because I do not use quorum queues. I use classic non-replicated queues that are reachable from the 2 RabbitMQ nodes.

Steps
You can use this code to reproduce the issue:

- Install RabbitMQ cluster with 2 nodes (NodeA, NodeB).
- Create RabbitClient.java java file containing source below. This program will 
    - create connection to one any available RabbitMQ node
    - create channel
    - create exchange
    - create 2000 queues (I use at least 2000 queues otherwise I can't reproduce the issue)
    - create bindings for each queue
    - create consumer on each queue
    - If the connection is lost, then we restart all steps
- Compile the java file:
    => javac -cp amqp-client-5.15.0.jar RabbitClient.java
- Run java program:
    => java -classpath lib/amqp-client-5.15.0.jar:lib/slf4j-api-1.5.6.jar:lib/slf4j-jdk14-1.5.6.jar:. RabbitClient NodeA,NodeB / user password
- I can then publish messages on any queues (using routing keys) through RabbitMQ web interface management.
- Stop the RabbitMQ node on which the queues are created, the java client will recreate queues and bindings on the available RabbitMQ node
- Issue: All queues and bindings are created but some bindings doesn’t work.
    For my last test:
    - queues.0 to queues.99 not work, I publish message on RabbitMQ interface => error: Message published, but not routed.
    - queues.100 to queues.1999 work correctly.
    - Note that the issue occurs only when I publish message on exchange with routing key. It doesn't occur if I publish the message directly on the queue. 
    - Note that this issue occurs also with 3 RabbitMQ nodes.




import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Address;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.ConsumerCancelledException;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.ShutdownListener;
import com.rabbitmq.client.ShutdownSignalException;
import com.rabbitmq.utility.Utility;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class RabbitClient {
    private static Address[] addresses = null;
    private static ConnectionFactory connectionFactory;
    private static Connection connection = null;

    public static void main(String[] args) {
        String address = args[0];
        String virtualHost = args[1];
        String username = args[2];
        String password = args[3];
        connectionFactory = new ConnectionFactory();
        connectionFactory.setVirtualHost(virtualHost);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setAutomaticRecoveryEnabled(false);
        connectionFactory.setRequestedHeartbeat(10);
        addresses = Address.parseAddresses(address);

        while (true) {
            Channel channel = null;
            Consumer consumer = null;
            try {
                if (connection == null) {
                    try {
                        connection = connectionFactory.newConnection(addresses);
                        connection.addShutdownListener(new RMQShutdownListener());
                    } catch (TimeoutException e) {
                        System.out.println("error connection: " + e);
                    }
                }
                channel = connection.createChannel();
                channel.basicQos(1, true);
                consumer = new Consumer(channel);
                System.out.println("channel opened to " + channel.getConnection().getAddress());

                String exchangeName = "exchange.test";
                channel.exchangeDeclare(exchangeName, "topic", false, false, null);
                System.out.println("exchange created: " + exchangeName);

                System.out.println("create queues...");
                for ( int i = 0 ; i < 2000; i++ ) {
                    String queueName = "queue.test."+i;
                    channel.queueDeclare(queueName, false, false, false, null);
                    channel.queueBind(queueName, exchangeName, queueName, null);
                    channel.basicConsume(queueName, false, consumer);
                }
                System.out.println("queues created");

                while (true) {
                    String delivery = consumer.nextDelivery();
                    if (delivery != null) {
                        System.out.println("delivery: " + delivery);
                    }
                }
            } catch (Exception e) {
                System.out.println("error: " + e);
            } catch (OutOfMemoryError e) {
                System.out.println("memory error: " + e);
            }

            try {
                if (channel != null) {
                    channel.abort();
                }
            } catch (IOException e) {
                System.out.println("channel close error: " + e);
            }
            System.out.println("channel closed");
        }
    }

    public static class RMQShutdownListener implements ShutdownListener {
        @Override
        public void shutdownCompleted(ShutdownSignalException e) {
            if (!e.isInitiatedByApplication()) {
                System.out.println("disconnection from broker" + connection.getAddress());
                connection.abort(AMQP.CHANNEL_ERROR, "connection to broker has been lost", 0);
            }
            connection = null;
        }
    }

    public static class Consumer extends DefaultConsumer {
        private final BlockingQueue<String> queue;
        private volatile ShutdownSignalException shutdown;

        public Consumer(Channel channel) {
            super(channel);
            this.queue = new LinkedBlockingQueue<>();
        }

        public String nextDelivery() throws InterruptedException, ShutdownSignalException, ConsumerCancelledException {
            String delivery = queue.poll(100, TimeUnit.MILLISECONDS);
            if (delivery == null) {
                checkShutdown();
            }
            return delivery;
        }

        @Override
        public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
            checkShutdown();
            queue.add(new String(body, StandardCharsets.UTF_8));
            this.getChannel().basicAck(envelope.getDeliveryTag(), false);
        }

        @Override
        public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) {
            shutdown = sig;
        }

        private void checkShutdown() {
            if (shutdown != null)
                throw Utility.fixStackTrace(shutdown);
        }
    }
}

[acogoluegnes]

Thanks for the follow-up, I managed to reproduce the problem locally, with a 2-node cluster. You are using transient queues (which are deprecated and will be removed in RabbitMQ 4.0 BTW) and this may be part of the problems.

We suspect there's a race condition during the clean-up of the binding/queues that are on the first node and the re-creation of the exact same bindings/queues on the second node.

While we're investigating more there are some workarounds you can try:

• wait a few seconds before re-connecting (I tested, it worked). The Java client supports automatic connection recovery which tries to recover the connection after 5 seconds by default.
• use another kind of queue if it matches your use case: transient exclusive queues, durable non-replicated queues (not ideal for availability because the initial node needs to be back online), quorum queues.

[ridi2183]

Thank your for your answers and suggestions, but:

• Wait a few seconds 5 seconds before re-connecting works correclty but it's depends on the number of queues.

• The use of another type of queues is not suitable for the moment for other constraints.

Are you considering making a solution/fix for transient queues or it is not relevant ?

[michaelklishin]

There is natural race condition between what RabbitMQ nodes do (remove transient queues) and what reconnecting clients do (re-declare them).

There is no fundamental solution besides using durable entities or not using automatic topology recovery.

[acogoluegnes]

5 seconds is just a suggestion and the connection recovery interval default for most of the client libraries, you can increase it if it works better for you.

[ridi2183]

Is there a way to know or detect via the management API, for example, which bindings are not working correctly ?

[michaelklishin]

Unroutable message handling?