ActiveMQ fail-over of producer and consumer with a shared directory doesn't happen

Question

We have two ActiveMQ(version 5.10.0) instances running and I am using the shared storage to achieve HA. However I am unable to see failover happening for the producer and consumer(s).

ActiveMQ broker-1 runs on IP1 and broker-2 on IP2 And under the activemq.xml of configuration I have modified persistence adapter to use a shared directory which is present on IP1.

<persistenceAdapter>
  <kahaDB directory="\\IP1\shared-directory\for activemq\data"/>
</persistenceAdapter>

Both in producer and consumer sides I am using following JNDI configurations to get the connections and build sessions,etc.

jndi.properties

java.naming.factory.initial = ..........ActiveMQInitialContextFactory
java.naming.provider.url = failover:(tcp://IP1:61616,tcp://IP2:61616)?randomize=false
connectionFactoryNames = myConnectionFactory
queue.requestQ = my.RequestQ

Interesting part is :

When I start this broker pair, I see that one of the brokers becomes master. When I start the producer, which puts the message on the Q (say producer has put 100 messages on the Q). While my producer is still running; I shutdown master broker, hence slave broker acquires the file-lock and becomes master.When I open the webconsole I see that 100 messages are still there on the Q. Even though producer is running it no longer puts any messages on this Q.

Similar to this for the consumers also. Consumer was picking messages from the Q, this Q has say 100 messages unconsumed when master failed, now master goes down, slave becomes master, I see 100 messages are still unconsumed, but the consumer does not pick any message from the Q.

I waited them to failover for a long time.(>10 mins.) Can any one please suggest what configuration am I missing ?

I am copy pasting producer and consumer as is (I've copied this from ActiveMQ in action book with minor modifications).

Producer

public class Producer {

    private static String brokerURL = "failover:(tcp://IP1:3389,tcp://IP2:3389)";
    private static transient ConnectionFactory factory;
    private transient Connection connection;
    private transient Session session;
    private transient MessageProducer producer;

    private static int count = 10;
    private static int total;
    private static int id = 1000000;
    private String jobs[] = new String[] { "suspend", "delete" };

    public Producer() throws JMSException {
        factory = new ActiveMQConnectionFactory(brokerURL);
        connection = factory.createConnection();
        connection.start();
        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        producer = session.createProducer(null);
    }

    public void close() throws JMSException {
        if (connection != null) {
            connection.close();
        }
    }

    public static void main(String[] args) throws JMSException {
        Producer producer = new Producer();
        while (total < 1000) {
            for (int i = 0; i < count; i++) {
                producer.sendMessage();
            }
            total += count;
            System.out.println("Sent '" + count + "' of '" + total
                    + "' job messages");
            try {
                Thread.sleep(1000);
            } catch (InterruptedException x) {
            }
        }
        producer.close();

    }

    public void sendMessage() throws JMSException {
        int idx = 0;
        while (true) {
            idx = (int) Math.round(jobs.length * Math.random());
            if (idx < jobs.length) {
                break;
            }
        }
        String job = jobs[idx];
        Destination destination = session.createQueue("JOBS." + job);
        Message message = session.createObjectMessage(id++);
        System.out.println("Sending: id: "
                + ((ObjectMessage) message).getObject() + " on queue: "
                + destination);
        producer.send(destination, message);
    }
}

Consumer

public class Consumer {

    private static String brokerURL = "failover:(tcp://IP1:3389,tcp://IP2:3389)";
    private static transient ConnectionFactory factory;
    private transient Connection connection;
    private transient Session session;

    private String jobs[] = new String[] { "suspend", "delete" };

    public Consumer() throws JMSException {
        factory = new ActiveMQConnectionFactory(brokerURL);
        connection = factory.createConnection();
        connection.start();
        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    }

    public void close() throws JMSException {
        if (connection != null) {
            connection.close();
        }
    }

    public static void main(String[] args) throws JMSException {
        Consumer consumer = new Consumer();
        for (String job : consumer.jobs) {
            Destination destination = consumer.getSession().createQueue(
                    "JOBS." + job);
            MessageConsumer messageConsumer = consumer.getSession()
                    .createConsumer(destination);
            messageConsumer.setMessageListener(new Listener(job));
        }
    }

    public Session getSession() {
        return session;
    }

}

Just one more thing: I am more interested in consumer failover than producer. One more observation is : Consumer stops and comes to the command prompt abruptly.

Thank you. -JE