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@904270 13f79535-47bb-0310-9956-ffa450edef68

8 files changed, 87 insertions, 0 deletions

diff --git a/qpid/cpp/src/qpid/agent/ManagementAgentImpl.h b/qpid/cpp/src/qpid/agent/ManagementAgentImpl.h
index a876496e98..b1efa1809b 100644
--- a/qpid/cpp/src/qpid/agent/ManagementAgentImpl.h
+++ b/qpid/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/qpid/cpp/src/qpid/cluster/Connection.cpp b/qpid/cpp/src/qpid/cluster/Connection.cpp
index 3f37d63255..c1c9c02611 100644
--- a/qpid/cpp/src/qpid/cluster/Connection.cpp
+++ b/qpid/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/qpid/cpp/src/qpid/cluster/Connection.h b/qpid/cpp/src/qpid/cluster/Connection.h
index ca9b27ef3f..7345378130 100644
--- a/qpid/cpp/src/qpid/cluster/Connection.h
+++ b/qpid/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/qpid/cpp/src/qpid/cluster/UpdateClient.cpp b/qpid/cpp/src/qpid/cluster/UpdateClient.cpp
index d4bd4da7f8..3fba194d44 100644
--- a/qpid/cpp/src/qpid/cluster/UpdateClient.cpp
+++ b/qpid/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/qpid/cpp/src/qpid/cluster/UpdateClient.h b/qpid/cpp/src/qpid/cluster/UpdateClient.h
index 29ef5f9df2..7407b7450b 100644
--- a/qpid/cpp/src/qpid/cluster/UpdateClient.h
+++ b/qpid/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/qpid/cpp/src/qpid/management/ManagementAgent.h b/qpid/cpp/src/qpid/management/ManagementAgent.h
index 3dea8ce3c7..ca8bfe97ed 100644
--- a/qpid/cpp/src/qpid/management/ManagementAgent.h
+++ b/qpid/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/qpid/cpp/xml/cluster.xml b/qpid/cpp/xml/cluster.xml
index 8513476d06..569cebaf14 100644
--- a/qpid/cpp/xml/cluster.xml
+++ b/qpid/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>
diff --git a/qpid/specs/management-schema.xml b/qpid/specs/management-schema.xml
index 1e345b5ea5..b2e732e9e3 100644
--- a/qpid/specs/management-schema.xml
+++ b/qpid/specs/management-schema.xml
@@ -326,6 +326,27 @@
     <method name="close"/>
   </class>
 
+  <!--
+  ===============================================================
+  ManagementSetupState
+  ===============================================================
+
+     This thing is used during cluster recovery operations (and maybe
+  eventually elsewhere) to transmit assorted state from one broker to
+  another.  At present, the two data propagated are the object number
+  counter and boot sequence, both of which are used for creating
+  object ids for newly-created objects.
+
+    -->
+  <class name="ManagementSetupState">
+    <!-- for reasons that aren't clear (to me, anyhow) you have to say
+    access="RO" to get accessor methods defined.  RC or RW don't do
+    it.  Probably this is documented someplace, but I couldn't find
+    it. -jrd -->
+    <property name="objectNum" type="uint64" access="RO"/>
+    <property name="bootSequence" type="uint16" access="RO"/>
+  </class>
+
   <eventArguments>
     <arg name="altEx"   type="sstr"   desc="Name of the alternate exchange"/>
     <arg name="args"    type="map"    desc="Supplemental arguments or parameters supplied"/>