[rabbitmq-discuss] Low disk space causes RabbitMQ to loose messages instead of blocking the connection

[Joe]

I have a java application that publishes messages to a RabbitMQ server. 
When the available disk space drops below rabbit's low watermark I get
unexpected behavior.  

The expected behavior is that  the connection will become blocking
<http://www.rabbitmq.com/memory.html>  , making my application hang on the
call to /Channel.basicPublish/.

The actual behavior is that the connection appears to be blocking in the
management console, but calls to /Channel.basicPublish/ return with no
errors and the messages that were supposed to be published are lost.  
This behavior undermines the most important feature of RabbitMQ, which is
robustness.

Below is a minimal version of my application for testing. All it does is
publish a message every second with an incrementing index (1, 2, 3, ...).
The messages are received just fine by the RabbitMQ server, until I set the
low watermark to a very high value, by putting the following line in the
/rabbitmq.config/ file:

    [    
      {rabbit, [{disk_free_limit, 60000000000}]}
    ].

After restarting the server, I get a low disk space notification in the
management console, the connection is marked as 'blocking', and no more
messages are received by the server. However, the application keeps running
and sending messages as if nothing is wrong. When I reduce the watermark
back to a normal value messages are received by the server again, but all
the messages that were sent while the connection was blocking are lost.

 - *Am I doing something wrong?*
 - *Is this a bug in RabbitMQ?*
 - *If so, is there a workaround?*

OS: Windows 8 64bit  
RabbitMQ server version: 3.1.1  
RabbitMQ Java client version: 3.1.0  

Test application code:

    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.MessageProperties;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import java.io.IOException;
    
    public class Main {
    
        private final static Logger logger =
LoggerFactory.getLogger(Main.class);
        private final static String QUEUE_NAME = "testQueue";
    
        private static Channel channel = null;
    
        private static void connectToRabbitMQ() throws IOException {
            ConnectionFactory factory = new ConnectionFactory();
            Connection connection = factory.newConnection();
            channel = connection.createChannel();
            channel.queueDeclare(
                    QUEUE_NAME,
                    true,       // Durable - survive a server restart
                    false,      // Not exclusive to this connection
                    false,      // Do not autodelete when no longer in use
                    null        // Arguments
            );
        }
    
        private static void disposeChannel()
        {
            if (channel == null) {
                return;
            }
            try {
                channel.close();
            } catch (Exception e) {
            } finally {
                channel = null;
            }
        }
    
        public static void main(String[] args) throws Exception {
    
            boolean interrupted = false;
            int messageNumber = 1;
    
            while (!interrupted) {
                byte[] message = Integer.toString(messageNumber).getBytes();
                try {
                    if (channel == null) {
                        connectToRabbitMQ();
                    }
                    channel.basicPublish(
                            "",
                            QUEUE_NAME,
                            MessageProperties.MINIMAL_PERSISTENT_BASIC,
                            message
                    );
                    logger.info("Published message number {}",
messageNumber);
                    messageNumber++;
                } catch (Exception e) {
                    logger.info("Unable to connect to RabbitMQ...");
                    disposeChannel();
                }
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    logger.info("Interrupted");
                    interrupted = true;
                }
            }
        }
    }



--
View this message in context: http://rabbitmq.1065348.n5.nabble.com/Low-disk-space-causes-RabbitMQ-to-loose-messages-instead-of-blocking-the-connection-tp27157.html
Sent from the RabbitMQ mailing list archive at Nabble.com.