[rabbitmq-discuss] Cluster Pathology

[Drew Smathers]

On Thu, Feb 12, 2009 at 6:17 AM, Ben Hood <0x6e6562 at gmail.com> wrote:
> Drew,
>
> On Thu, Feb 12, 2009 at 5:18 AM, Drew Smathers <drew.smathers at gmail.com> wrote:
>> Our experience has been the messages are lost (they're also marked
>> persistent via delivery_mode=2 and queues are durable).
>
> Interesting. And you definitely have at least one route to at least on
> queue for each routing key?
>
> Ben
>

Yes, the publisher in our case preemptively declares and binds queues
- to ensure messages are not lost.  (The consumers also make the same
queue declaration).  Below I've included a stripped down example of a
publisher and consumer (using py-amqplib) which can be used to
reproduce the issue.

Steps to reproduce:

1. run publisher and consumer against one node to ensure queue is created there:

 $ python publisher.py hostB 5
 $ python consumer.py hostB # CTL-C after receiving 5 messages

2. run publisher/consumer against other node - hostA

 $ python publisher.py hostA 20
 $ python consumer.py hostA

3. Before publisher from step 2 has finished, bring down rabbitmq on hostB

  hostB $ rabbitmqctl stop

4. After publisher from step 2 has finished, restart consumer:

  $ python consumer.py hostA

Notice messages delivered after hostB was brought down were not delivered.

==== publisher.py ==============
import sys
from time import sleep, time
from amqplib import client_0_8 as amqp

host = sys.argv[1]
message_count = int(sys.argv[2])

SLEEP = 2
EXCHANGE = 'publisher_test'
QUEUE = 'queue1'

connection = amqp.Connection(host, userid='guest', password='guest',
        insist=True, connect_timeout=5)
channel = connection.channel()
channel.access_request('/data', active=True, read=True, write=True)
channel.exchange_declare(EXCHANGE, 'direct',
            auto_delete=False, durable=True)
channel.queue_declare(QUEUE, auto_delete=False, durable=True)
channel.queue_bind(QUEUE, EXCHANGE, routing_key='messages')

for ct in range(message_count):
    message = amqp.Message('%d-%d' % (int(1000 * time()), ct),
                           content_type='text/plain', delivery_mode=2)
    channel.basic_publish(message, EXCHANGE,
        routing_key='messages', mandatory=True)
    print 'published message', message.body
    sleep(SLEEP)

channel.close()
connection.close()


==== python consumer.py  =======================

import sys
from amqplib import client_0_8 as amqp

host = sys.argv[1]

EXCHANGE = 'publisher_test'
QUEUE = 'queue1'

connection = amqp.Connection(host, userid='guest', password='guest',
        insist=True, connect_timeout=5)
channel = connection.channel()
channel.access_request('/data', active=True, read=True, write=True)
channel.exchange_declare(EXCHANGE, 'direct',
            auto_delete=False, durable=True)
channel.queue_declare(QUEUE, auto_delete=False, durable=True)
channel.queue_bind(QUEUE, EXCHANGE, routing_key='messages')

def handle_message(message):
    print 'got message', message.body
    channel.basic_ack(message.delivery_tag)

tag = channel.basic_consume(
    QUEUE, callback=handle_message)
while channel.callbacks:
    try:
        channel.wait()
    except KeyboardInterrupt:
        break

channel.close()
connection.close()