QPID-2357 Broker boot sequence doesn't synchronize when clustered - patch from John Dunning

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

7 files changed, 66 insertions, 0 deletions

diff --git a/cpp/src/qpid/agent/ManagementAgentImpl.h b/cpp/src/qpid/agent/ManagementAgentImpl.h
index a876496e98..b1efa1809b 100644
--- a/cpp/src/qpid/agent/ManagementAgentImpl.h
+++ b/cpp/src/qpid/agent/ManagementAgentImpl.h
@@ -82,6 +82,13 @@ class ManagementAgentImpl : public ManagementAgent, public client::MessageListen
     uint16_t getInterval() { return interval; }
     void periodicProcessing();
 
+    // these next are here to support the hot-wiring of state between clustered brokers
+    uint64_t getNextObjectId(void) { return nextObjectId; }
+    void setNextObjectId(uint64_t o) { nextObjectId = o; }
+
+    uint16_t getBootSequence(void) { return bootSequence; }
+    void setBootSequence(uint16_t b) { bootSequence = b; }
+
   private:
 
     struct SchemaClassKey {
diff --git a/cpp/src/qpid/cluster/Connection.cpp b/cpp/src/qpid/cluster/Connection.cpp
index 3f37d63255..c1c9c02611 100644
--- a/cpp/src/qpid/cluster/Connection.cpp
+++ b/cpp/src/qpid/cluster/Connection.cpp
@@ -513,5 +513,19 @@ void Connection::managementSchema(const std::string& data) {
     QPID_LOG(debug, cluster << " updated management schemas");
 }
 
+//
+// This is the handler for incoming managementsetup messages.
+//
+void Connection::managementSetupState(uint64_t objectNum, uint16_t bootSequence)
+{
+    QPID_LOG(debug, "Running managementsetup state handler, new objectnum "
+	     << objectNum << " seq " << bootSequence);
+    management::ManagementAgent* agent = cluster.getBroker().getManagementAgent();
+    if (!agent)
+        throw Exception(QPID_MSG("Management schema update but no management agent."));
+    agent->setNextObjectId(objectNum);
+    agent->setBootSequence(bootSequence);
+}
+
 }} // Namespace qpid::cluster
 
diff --git a/cpp/src/qpid/cluster/Connection.h b/cpp/src/qpid/cluster/Connection.h
index ca9b27ef3f..7345378130 100644
--- a/cpp/src/qpid/cluster/Connection.h
+++ b/cpp/src/qpid/cluster/Connection.h
@@ -164,6 +164,7 @@ class Connection :
 
     void addQueueListener(const std::string& queue, uint32_t listener);
     void managementSchema(const std::string& data);
+    void managementSetupState(uint64_t objectNum, uint16_t bootSequence);
 
     uint32_t getSsf() const { return connectionCtor.ssf; }
 
diff --git a/cpp/src/qpid/cluster/UpdateClient.cpp b/cpp/src/qpid/cluster/UpdateClient.cpp
index d4bd4da7f8..3fba194d44 100644
--- a/cpp/src/qpid/cluster/UpdateClient.cpp
+++ b/cpp/src/qpid/cluster/UpdateClient.cpp
@@ -127,6 +127,17 @@ void UpdateClient::update() {
     QPID_LOG(debug, updaterId << " updating state to " << updateeId
              << " at " << updateeUrl);
     Broker& b = updaterBroker;
+
+    //
+    // Bash the state of the slave into conformance with ours.  The
+    // goal here is to get his state arranged so as to mimic our
+    // state, w/r/t object ID creation.  Currently, that means that we
+    // propagate our boot seq and object UID counter to him so that
+    // subsequently created objects on his side will track what's on
+    // our side.
+    //
+    updateManagementSetupState(b);
+
     b.getExchanges().eachExchange(boost::bind(&UpdateClient::updateExchange, this, _1));
     b.getQueues().eachQueue(boost::bind(&UpdateClient::updateNonExclusiveQueue, this, _1));
 
@@ -173,6 +184,22 @@ template <class T> std::string encode(const T& t) {
 }
 } // namespace
 
+//
+// Propagate the management setup state block, currently consisting of
+// object number counter and boot sequence counter, to the slave.
+//
+void UpdateClient::updateManagementSetupState(Broker & b)
+{
+    qmf::org::apache::qpid::broker::ManagementSetupState mss(b.getManagementAgent(), 0);
+    mss.set_objectNum(b.getManagementAgent()->getNextObjectId());
+    mss.set_bootSequence(b.getManagementAgent()->getBootSequence());
+
+    QPID_LOG(debug, updaterId << " updating management-setup-state " << mss.get_objectNum() 
+	     << " " << mss.get_bootSequence() << "\n");
+
+    ClusterConnectionProxy(session).managementSetupState(mss.get_objectNum(), mss.get_bootSequence());
+}
+
 void UpdateClient::updateExchange(const boost::shared_ptr<Exchange>& ex) {
     QPID_LOG(debug, updaterId << " updating exchange " << ex->getName());
     ClusterConnectionProxy(session).exchange(encode(*ex));
diff --git a/cpp/src/qpid/cluster/UpdateClient.h b/cpp/src/qpid/cluster/UpdateClient.h
index 29ef5f9df2..7407b7450b 100644
--- a/cpp/src/qpid/cluster/UpdateClient.h
+++ b/cpp/src/qpid/cluster/UpdateClient.h
@@ -29,6 +29,7 @@
 #include "qpid/client/AsyncSession.h"
 #include "qpid/broker/SemanticState.h"
 #include "qpid/sys/Runnable.h"
+#include "qmf/org/apache/qpid/broker/ManagementSetupState.h"
 #include <boost/shared_ptr.hpp>
 
 
@@ -97,6 +98,7 @@ class UpdateClient : public sys::Runnable {
     void updateConsumer(const broker::SemanticState::ConsumerImpl::shared_ptr&);
     void updateQueueListeners(const boost::shared_ptr<broker::Queue>&);
     void updateQueueListener(std::string& q, const boost::shared_ptr<broker::Consumer>& c);
+    void updateManagementSetupState(broker::Broker & b);
 
     Numbering<broker::SemanticState::ConsumerImpl::shared_ptr> consumerNumbering;
     MemberId updaterId;
diff --git a/cpp/src/qpid/management/ManagementAgent.h b/cpp/src/qpid/management/ManagementAgent.h
index 3dea8ce3c7..ca8bfe97ed 100644
--- a/cpp/src/qpid/management/ManagementAgent.h
+++ b/cpp/src/qpid/management/ManagementAgent.h
@@ -104,6 +104,13 @@ public:
     /** Decode a serialized schemas and add to my schema cache */
     void importSchemas(framing::Buffer& inBuf);
 
+    // these are in support of the managementSetup-state stuff, for synch'ing clustered brokers
+    uint64_t getNextObjectId(void) { return nextObjectId; }
+    void setNextObjectId(uint64_t o) { nextObjectId = o; }
+
+    uint16_t getBootSequence(void) { return bootSequence; }
+    void setBootSequence(uint16_t b) { bootSequence = b; }
+
 private:
     //  Storage for tracking remote management agents, attached via the client
     //  management agent API.
diff --git a/cpp/xml/cluster.xml b/cpp/xml/cluster.xml
index 8513476d06..569cebaf14 100644
--- a/cpp/xml/cluster.xml
+++ b/cpp/xml/cluster.xml
@@ -237,5 +237,13 @@
     <control name="management-schema" code="0x35">
       <field name="data" type="vbin32"/>
     </control>
+
+    <!-- added by jrd.  propagate a management-setup-state widget -->
+    <control name="management-setup-state" code="0x36">
+      <field name="objectNum" type="uint64"/>
+      <field name="bootSequence" type="uint16"/>
+    </control>
+
+
   </class>
 </amqp>