QPID-4325: HA Starting from persistent store

When re-starting a persistent HA cluster, the broker that becomes primary should
keep its store data while all the backup brokers should discard their store data
and catch up from the primary. Backups cannot simply use their own stores
because sequence numbers of stored messages will not match on all brokers. The
backup erases individual queues and exchanges as the catch-up process gets to
them.

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@1390123 13f79535-47bb-0310-9956-ffa450edef68

7 files changed, 71 insertions, 38 deletions

diff --git a/cpp/src/qpid/ha/AlternateExchangeSetter.h b/cpp/src/qpid/ha/AlternateExchangeSetter.h
index 08690e68bc..1878939aad 100644
--- a/cpp/src/qpid/ha/AlternateExchangeSetter.h
+++ b/cpp/src/qpid/ha/AlternateExchangeSetter.h
@@ -43,12 +43,14 @@ class AlternateExchangeSetter
 
     AlternateExchangeSetter(broker::ExchangeRegistry& er) : exchanges(er) {}
 
+    /** If altEx is already known, call setter(altEx) now else save for later */
     void setAlternate(const std::string& altEx, const SetFunction& setter) {
         broker::Exchange::shared_ptr ex = exchanges.find(altEx);
         if (ex) setter(ex);     // Set immediately.
         else setters.insert(Setters::value_type(altEx, setter)); // Save for later.
     }
 
+    /** Add an exchange and call any setters that are waiting for it. */
     void addExchange(boost::shared_ptr<broker::Exchange> exchange) {
         // Update the setters for this exchange
         std::pair<Setters::iterator, Setters::iterator> range = setters.equal_range(exchange->getName());
diff --git a/cpp/src/qpid/ha/BrokerReplicator.cpp b/cpp/src/qpid/ha/BrokerReplicator.cpp
index 0b6280da6d..2f0d304686 100644
--- a/cpp/src/qpid/ha/BrokerReplicator.cpp
+++ b/cpp/src/qpid/ha/BrokerReplicator.cpp
@@ -41,6 +41,7 @@
 #include "qmf/org/apache/qpid/broker/EventQueueDelete.h"
 #include "qmf/org/apache/qpid/broker/EventSubscribe.h"
 #include "qmf/org/apache/qpid/ha/EventMembersUpdate.h"
+#include <boost/bind.hpp>
 #include <algorithm>
 #include <sstream>
 #include <iostream>
@@ -313,11 +314,11 @@ void BrokerReplicator::doEventQueueDeclare(Variant::Map& values) {
         // The queue was definitely created on the primary.
         if (broker.getQueues().find(name)) {
             QPID_LOG(warning, logPrefix << "Replacing exsiting queue: " << name);
-            broker.getQueues().destroy(name);
+            broker.deleteQueue(name, userId, remoteHost);
             stopQueueReplicator(name);
         }
         settings.populate(args, settings.storeSettings);
-        std::pair<boost::shared_ptr<Queue>, bool> result =
+        CreateQueueResult result =
             broker.createQueue(
                 name,
                 settings,
@@ -325,7 +326,7 @@ void BrokerReplicator::doEventQueueDeclare(Variant::Map& values) {
                 values[ALTEX].asString(),
                 userId,
                 remoteHost);
-        assert(result.second);  // Should be true since we destroyed existing queue above
+        assert(result.second);
         startQueueReplicator(result.first);
     }
 }
@@ -361,12 +362,14 @@ void BrokerReplicator::doEventExchangeDeclare(Variant::Map& values) {
         // If we already have a exchange with this name, replace it.
         // The exchange was definitely created on the primary.
         if (broker.getExchanges().find(name)) {
-            broker.getExchanges().destroy(name);
+            broker.deleteExchange(name, userId, remoteHost);
             QPID_LOG(warning, logPrefix << "Replaced exsiting exchange: " << name);
         }
-        boost::shared_ptr<Exchange> exchange =
-            createExchange(name, values[EXTYPE].asString(), values[DURABLE].asBool(), args, values[ALTEX].asString());
-        assert(exchange);
+        CreateExchangeResult result = createExchange(
+            name, values[EXTYPE].asString(), values[DURABLE].asBool(), args,
+            values[ALTEX].asString());
+        replicatedExchanges.insert(name);
+        assert(result.second);
     }
 }
 
@@ -380,6 +383,7 @@ void BrokerReplicator::doEventExchangeDelete(Variant::Map& values) {
     } else {
         QPID_LOG(debug, logPrefix << "Exchange delete event:" << name);
         broker.deleteExchange(name, userId, remoteHost);
+        replicatedExchanges.erase(name);
     }
 }
 
@@ -399,7 +403,7 @@ void BrokerReplicator::doEventBind(Variant::Map& values) {
         QPID_LOG(debug, logPrefix << "Bind event: exchange=" << exchange->getName()
                  << " queue=" << queue->getName()
                  << " key=" << key);
-        exchange->bind(queue, key, &args);
+        queue->bind(exchange, key, args);
     }
 }
 
@@ -454,14 +458,20 @@ void BrokerReplicator::doResponseQueue(Variant::Map& values) {
         return;
     string name(values[NAME].asString());
     QPID_LOG(debug, logPrefix << "Queue response: " << name);
+    if (broker.getQueues().find(name)) { // Already exists
+        if (findQueueReplicator(name))
+            return; // Already replicated
+        else {
+            QPID_LOG(debug, logPrefix << "Deleting queue to make way for replica: " << name);
+            broker.deleteQueue(name, userId, remoteHost);
+        }
+    }
     framing::FieldTable args;
     qpid::amqp_0_10::translate(argsMap, args);
-    boost::shared_ptr<Queue> queue =
+    CreateQueueResult result =
         createQueue(name, values[DURABLE].asBool(), values[AUTODELETE].asBool(), args,
                     getAltExchange(values[ALTEXCHANGE]));
-    // It is normal for the queue to already exist if we are failing over.
-    if (queue) startQueueReplicator(queue);
-    else QPID_LOG(debug, logPrefix << "Queue already replicated: " << name);
+    if (result.second) startQueueReplicator(result.first);
 }
 
 void BrokerReplicator::doResponseExchange(Variant::Map& values) {
@@ -469,13 +479,22 @@ void BrokerReplicator::doResponseExchange(Variant::Map& values) {
     if (!replicationTest.replicateLevel(argsMap)) return;
     string name = values[NAME].asString();
     QPID_LOG(debug, logPrefix << "Exchange response: " << name);
+    if (broker.getExchanges().find(name)) {
+        if (replicatedExchanges.find(name) != replicatedExchanges.end())
+            return; // Already replicated
+        else {
+            QPID_LOG(debug, logPrefix << "Deleting exchange to make way for replica: "
+                     << name);
+            broker.deleteExchange(name, userId, remoteHost);
+        }
+    }
     framing::FieldTable args;
     qpid::amqp_0_10::translate(argsMap, args);
-    boost::shared_ptr<Exchange> exchange = createExchange(
+    CreateExchangeResult result = createExchange(
         name, values[TYPE].asString(), values[DURABLE].asBool(), args,
         getAltExchange(values[ALTEXCHANGE]));
-    // It is normal for the exchange to already exist if we are failing over.
-    QPID_LOG_IF(debug, !exchange, logPrefix << "Exchange already replicated: " << name);
+    assert(result.second);
+    replicatedExchanges.insert(name);
 }
 
 namespace {
@@ -563,7 +582,7 @@ void BrokerReplicator::stopQueueReplicator(const std::string& name) {
     }
 }
 
-boost::shared_ptr<Queue> BrokerReplicator::createQueue(
+BrokerReplicator::CreateQueueResult BrokerReplicator::createQueue(
     const std::string& name,
     bool durable,
     bool autodelete,
@@ -572,7 +591,7 @@ boost::shared_ptr<Queue> BrokerReplicator::createQueue(
 {
     QueueSettings settings(durable, autodelete);
     settings.populate(arguments, settings.storeSettings);
-    std::pair<boost::shared_ptr<Queue>, bool> result =
+    CreateQueueResult result =
         broker.createQueue(
             name,
             settings,
@@ -580,24 +599,22 @@ boost::shared_ptr<Queue> BrokerReplicator::createQueue(
             string(), // Set alternate exchange below
             userId,
             remoteHost);
-    if (result.second) {
-        if (!alternateExchange.empty()) {
-            alternates.setAlternate(
-                alternateExchange, boost::bind(&Queue::setAlternateExchange, result.first, _1));
-        }
-        return result.first;
+
+    if (!alternateExchange.empty()) {
+        alternates.setAlternate(
+            alternateExchange, boost::bind(&Queue::setAlternateExchange, result.first, _1));
     }
-    else return  boost::shared_ptr<Queue>();
+    return result;
 }
 
-boost::shared_ptr<Exchange> BrokerReplicator::createExchange(
+BrokerReplicator::CreateExchangeResult BrokerReplicator::createExchange(
     const std::string& name,
     const std::string& type,
     bool durable,
     const qpid::framing::FieldTable& args,
     const std::string& alternateExchange)
 {
-    std::pair<boost::shared_ptr<Exchange>, bool> result =
+    CreateExchangeResult result =
         broker.createExchange(
             name,
             type,
@@ -606,15 +623,13 @@ boost::shared_ptr<Exchange> BrokerReplicator::createExchange(
             args,
             userId,
             remoteHost);
-    if (result.second) {
-        alternates.addExchange(result.first);
-        if (!alternateExchange.empty()) {
-            alternates.setAlternate(
-                alternateExchange, boost::bind(&Exchange::setAlternate, result.first, _1));
-        }
-        return result.first;
+
+    alternates.addExchange(result.first);
+    if (!alternateExchange.empty()) {
+        alternates.setAlternate(
+            alternateExchange, boost::bind(&Exchange::setAlternate, result.first, _1));
     }
-    else return  boost::shared_ptr<Exchange>();
+    return result;
 }
 
 bool BrokerReplicator::bind(boost::shared_ptr<Queue>, const string&, const framing::FieldTable*) { return false; }
@@ -623,4 +638,5 @@ bool BrokerReplicator::isBound(boost::shared_ptr<Queue>, const string* const, co
 
 string BrokerReplicator::getType() const { return QPID_CONFIGURATION_REPLICATOR; }
 
+
 }} // namespace broker
diff --git a/cpp/src/qpid/ha/BrokerReplicator.h b/cpp/src/qpid/ha/BrokerReplicator.h
index 69653b876a..5c8a983d45 100644
--- a/cpp/src/qpid/ha/BrokerReplicator.h
+++ b/cpp/src/qpid/ha/BrokerReplicator.h
@@ -78,6 +78,8 @@ class BrokerReplicator : public broker::Exchange,
 
   private:
     typedef boost::shared_ptr<QueueReplicator> QueueReplicatorPtr;
+    typedef std::pair<boost::shared_ptr<broker::Queue>, bool> CreateQueueResult;
+    typedef std::pair<boost::shared_ptr<broker::Exchange>, bool> CreateExchangeResult;
 
     void initializeBridge(broker::Bridge&, broker::SessionHandler&);
 
@@ -98,14 +100,14 @@ class BrokerReplicator : public broker::Exchange,
     void startQueueReplicator(const boost::shared_ptr<broker::Queue>&);
     void stopQueueReplicator(const std::string& name);
 
-    boost::shared_ptr<broker::Queue> createQueue(
+    CreateQueueResult createQueue(
         const std::string& name,
         bool durable,
         bool autodelete,
         const qpid::framing::FieldTable& arguments,
         const std::string& alternateExchange);
 
-    boost::shared_ptr<broker::Exchange> createExchange(
+    CreateExchangeResult createExchange(
         const std::string& name,
         const std::string& type,
         bool durable,
@@ -121,6 +123,8 @@ class BrokerReplicator : public broker::Exchange,
     bool initialized;
     AlternateExchangeSetter alternates;
     qpid::Address primary;
+    typedef std::set<std::string> StringSet;
+    StringSet replicatedExchanges; // exchanges that have been replicated.
 };
 }} // namespace qpid::broker
 
diff --git a/cpp/src/qpid/ha/QueueGuard.cpp b/cpp/src/qpid/ha/QueueGuard.cpp
index b0ef167176..8852554d31 100644
--- a/cpp/src/qpid/ha/QueueGuard.cpp
+++ b/cpp/src/qpid/ha/QueueGuard.cpp
@@ -61,7 +61,10 @@ QueueGuard::QueueGuard(broker::Queue& q, const BrokerInfo& info)
     range = QueueRange(q);
 }
 
-QueueGuard::~QueueGuard() { cancel(); }
+QueueGuard::~QueueGuard() {
+    QPID_LOG(debug, logPrefix << "Cancelled");
+    cancel();
+}
 
 // NOTE: Called with message lock held.
 void QueueGuard::enqueued(const Message& m) {
@@ -97,7 +100,6 @@ void QueueGuard::completeRange(Delayed::iterator begin, Delayed::iterator end) {
 }
 
 void QueueGuard::cancel() {
-    QPID_LOG(debug, logPrefix << "Cancelled");
     queue.removeObserver(observer);
     Delayed removed;
     {
diff --git a/cpp/src/qpid/ha/QueueReplicator.cpp b/cpp/src/qpid/ha/QueueReplicator.cpp
index 82daad9d9c..07c94ad261 100644
--- a/cpp/src/qpid/ha/QueueReplicator.cpp
+++ b/cpp/src/qpid/ha/QueueReplicator.cpp
@@ -67,6 +67,7 @@ QueueReplicator::QueueReplicator(HaBroker& hb,
       logPrefix("Backup queue "+q->getName()+": "),
       queue(q), link(l), brokerInfo(hb.getBrokerInfo())
 {
+    args.setString(QPID_REPLICATE, printable(NONE).str());
     Uuid uuid(true);
     bridgeName = replicatorName(q->getName()) + std::string(".") + uuid.str();
 }
diff --git a/cpp/src/qpid/ha/ReplicationTest.cpp b/cpp/src/qpid/ha/ReplicationTest.cpp
index 88a969dbfd..1dd32262a0 100644
--- a/cpp/src/qpid/ha/ReplicationTest.cpp
+++ b/cpp/src/qpid/ha/ReplicationTest.cpp
@@ -20,6 +20,7 @@
  */
 #include "ReplicationTest.h"
 #include "qpid/broker/Queue.h"
+#include "qpid/broker/Exchange.h"
 #include "qpid/framing/FieldTable.h"
 
 namespace qpid {
@@ -71,5 +72,10 @@ bool ReplicationTest::isReplicated(ReplicateLevel level, const broker::Queue& q)
     return isReplicated(level, q.getSettings().storeSettings, q.isAutoDelete(), q.hasExclusiveOwner());
 }
 
+bool ReplicationTest::isReplicated(ReplicateLevel level, const broker::Exchange& ex)
+{
+    return replicateLevel(ex.getArgs()) >= level;
+}
+
 
 }} // namespace qpid::ha
diff --git a/cpp/src/qpid/ha/ReplicationTest.h b/cpp/src/qpid/ha/ReplicationTest.h
index 9f6976a8e4..ab6b1a6bcc 100644
--- a/cpp/src/qpid/ha/ReplicationTest.h
+++ b/cpp/src/qpid/ha/ReplicationTest.h
@@ -30,6 +30,7 @@ namespace qpid {
 
 namespace broker {
 class Queue;
+class Exchange;
 }
 
 namespace framing {
@@ -59,6 +60,7 @@ class ReplicationTest
     bool isReplicated(ReplicateLevel level,
                       const framing::FieldTable& args, bool autodelete, bool exclusive);
     bool isReplicated(ReplicateLevel level, const broker::Queue&);
+    bool isReplicated(ReplicateLevel level, const broker::Exchange&);
   private:
     ReplicateLevel replicateDefault;
 };