[rabbitmq-discuss] Active-Active implementation issues

[Gomathi Nayagam]

I am trying to implement the active-active High availability Queue and i
have some issues in the implementation. The problem is when we consume
messages from the Mirrored queue if any one of the node is down, we can
able to consume the messages after the node failure. But the consumed
messages after the node failure are still available in the rabbitmq console
even after acking all of the messages.

Some information about the overview of the configuration to test
active-active queues

I have NODE1 in Host1 and NODE2 in Host2. Both of the nodes are in cluster.
I have a Queue named TestQ in NODE1, since i have made this queue (TestQ)
as Mirrored Queue through the policies it is available in Node2 which is
running in Host2.

I am passing the array of Address(Host1, Host2) to get the connection in my
consumer. the code snippet is

Address address1 = new Address(Host1, 5672);
Address address2 = new Address(Host2, 5672);

RabbitmqConsumer consumer = new RabbitmqConsumer(new Address[]{address1,
address2}, queue);

While getting the connection,
Connection conn = connectionFactory.newConnection(addresses);

My use case is as below.

I have 30 messages in TestQ and i can see it in the RabbitMQ console of
Host1,Host2 looks like below

Host1:
Name    Node
TestQ    rabbit at Node2

Host2:
Name    Node
TestQ    rabbit at Node2

In my consumer, I have implemented my own consumer as below.

class MyConsumer extends QueueingConsumer{

    Channel channel;
    public MyConsumer(Channel ch) {
        super(ch);
        this.channel = ch;
    }

    @Override
    public void handleCancel(String consumerTag)  {
        System.out.println("handleCancel :: consumertag "+consumerTag+"
channel opened "+channel.isOpen()+",channel "+channel+", rabbitmq consumer
ch "+RabbitmqConsumer.ch);

    }
}

If i run my consumer, it creates a channel with the reference of
AMQChannel(amqp://guest@Host1:5672/,1) and it consumes the messasges from
the TestQ. I have checked the consumer tag of the channel in both of the
host and both are same.

After some of the messages (assume 10 messages) are consumed, i explicitly
down the Node2 by giving the below command

node2$ ./rabbitmqctl stop_app

After the failure of Node2, i can see the print statements which i added in
the handleCancel() method and consumer consumes all of the messages with
out any error. The problem is the even though the remaining 20 messages are
consumed and acked in the consumer, but if i check it in the rabbitmq
console, the 20 messages are still exists in the TestQ.

What i have to do in this scenario to overcome this issue.
-------------- next part --------------
An HTML attachment was scrubbed...
URL: <http://lists.rabbitmq.com/pipermail/rabbitmq-discuss/attachments/20130701/24ad62a2/attachment.htm>