QPID-4085: HA message-loss race condition, handling replication event after response.

If the backup broker receives a declare event for a queue after receiving a
queue response for the same queue, it removes the queue and replaces it with the
new one from the reponse. Previously it did not remove the corresponding bridge
so things fail when we attempt to create it. Corrected to remove the bridge also.

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

1 files changed, 12 insertions, 7 deletions

diff --git a/qpid/cpp/src/qpid/ha/BrokerReplicator.cpp b/qpid/cpp/src/qpid/ha/BrokerReplicator.cpp
index 83c2eaa144..1fabff6a09 100644
--- a/qpid/cpp/src/qpid/ha/BrokerReplicator.cpp
+++ b/qpid/cpp/src/qpid/ha/BrokerReplicator.cpp
@@ -312,6 +312,7 @@ void BrokerReplicator::doEventQueueDeclare(Variant::Map& values) {
         if (broker.getQueues().find(name)) {
             QPID_LOG(warning, logPrefix << "Replacing exsiting queue: " << name);
             broker.getQueues().destroy(name);
+            stopQueueReplicator(name);
         }
         std::pair<boost::shared_ptr<Queue>, bool> result =
             broker.createQueue(
@@ -343,13 +344,7 @@ void BrokerReplicator::doEventQueueDelete(Variant::Map& values) {
     boost::shared_ptr<Queue> queue = broker.getQueues().find(name);
     if (queue && replicationTest.replicateLevel(queue->getSettings())) {
         QPID_LOG(debug, logPrefix << "Queue delete event: " << name);
-        boost::shared_ptr<QueueReplicator> qr = findQueueReplicator(name);
-        if (qr) {
-            qr->deactivate();
-            // QueueReplicator's bridge is now queued for destruction but may not
-            // actually be destroyed.
-            broker.getExchanges().destroy(qr->getName());
-        }
+        stopQueueReplicator(name);
         broker.deleteQueue(name, values[USER].asString(), values[RHOST].asString());
     }
 }
@@ -563,6 +558,16 @@ void BrokerReplicator::startQueueReplicator(const boost::shared_ptr<Queue>& queu
     }
 }
 
+void BrokerReplicator::stopQueueReplicator(const std::string& name) {
+    boost::shared_ptr<QueueReplicator> qr = findQueueReplicator(name);
+    if (qr) {
+        qr->deactivate();
+        // QueueReplicator's bridge is now queued for destruction but may not
+        // actually be destroyed.
+        broker.getExchanges().destroy(qr->getName());
+    }
+}
+
 bool BrokerReplicator::bind(boost::shared_ptr<Queue>, const string&, const framing::FieldTable*) { return false; }
 bool BrokerReplicator::unbind(boost::shared_ptr<Queue>, const string&, const framing::FieldTable*) { return false; }
 bool BrokerReplicator::isBound(boost::shared_ptr<Queue>, const string* const, const framing::FieldTable* const) { return false; }