CODEWITHSUNDEEP
javarabbitmqjms
Roman Goyenko
Roman Goyenko

Reputation: 7070

RabbitMQ not failing send when messages are rejected due to queue size

I am using RMQ and it's JMS client to publish messages to RMQ (this is a requirement i have, I can't use their java client instead of JMS client).

So, basically I do this: 

        RMQConnectionFactory factory = new RMQConnectionFactory() ;
        factory.setUsername(props.getProperty("rmq.username"));
        factory.setPassword(props.getProperty("rmq.password"));
        factory.setHost(props.getProperty("rmq.host"));
        factory.setVirtualHost(props.getProperty("rmq.virtualHost"));
        factory.setPort(Integer.parseInt(props.getProperty("rmq.port")));

        Connection connection = factory.createConnection();
        connection.start();
        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        String queueName = managerProps.getProperty("rmq.queue.name");
        Queue queue = session.createQueue(queueName);
        producer = session.createProducer(queue);

        TextMessage msg = session.createTextMessage(text);
        msg.setText(text);
        producer.send(msg);

I have a policy set up on RMQ overflow: reject-publish, so if it's over the limit RMQ is supposed to send a nack when the queue is full, but I don't seem to get it.

The question is - how do I determine if the message was rejected? I assume the producer.send(msg) to be synchronous and throw exception if the message is not published, but I don't get any exceptions, it just looks like everything got published.

JMS spec has a send(msg, CompletionListener) with a listener with two methods onCompletion and onException, but it doesn't look like RMQ JMS client implemented this method.

Is there another way to make sure that message made it through?

Upvotes: 0

Views: 1463

Answers (2)

Roman Goyenko
Roman Goyenko

Reputation: 7070

I did some digging, the CompletionListener is part of JMS 2.0 and RMQ only implements JMS 1.1, that's the reason it's not there.

But it looks like I can do something with transactions. I would need to change the code like this: 

    RMQConnectionFactory factory = new RMQConnectionFactory() ;
    // ... skipping the code here
    connection.start();
    // set session to be transacted 
    session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
    String queueName = managerProps.getProperty("rmq.queue.name");
    Queue queue = session.createQueue(queueName);
    producer = session.createProducer(queue);

    TextMessage msg = session.createTextMessage(text);
    msg.setText(text);
    producer.send(msg);
    // commit transaction
    session.commit();

This will work if the queue is not full, but will throw an exception after a rejected message with this:

Caused by: com.rabbitmq.client.ShutdownSignalException: channel error; protocol method: #method(reply-code=406, reply-text=PRECONDITION_FAILED - partial tx completion, class-id=90, method-id=20)

I can then catch the exception and do what I need to do to resend/save the message.

Upvotes: 0

menya
menya

Reputation: 1525

RabbitMQ use Publisher Confirms to guarantee that a message isn't lost, so if your Queue overflow behavior is reject-publish, the confirm channel will got a nack. It is also contains in many AMQP client.

But in JMS client, I have check the code in rabbitmq-jms-client, and no send implementaion contains CompletionListener. So if you want to enjoy reliable publish, please use AMQP client.

Upvotes: 1

Related Questions