qpid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From acon...@apache.org
Subject svn commit: r893175 - in /qpid/trunk/qpid/cpp: src/qpid/cluster/Connection.cpp src/qpid/cluster/Connection.h src/qpid/cluster/UpdateClient.cpp src/tests/cluster_test.cpp xml/cluster.xml
Date Tue, 22 Dec 2009 14:22:18 GMT
Author: aconway
Date: Tue Dec 22 14:22:17 2009
New Revision: 893175

URL: http://svn.apache.org/viewvc?rev=893175&view=rev
Log:
QPID-2296: Cluster errors when using acquire-mode-not-acquired

Replicate consumer's queue position to new cluster nodes.

Modified:
    qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp
    qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.h
    qpid/trunk/qpid/cpp/src/qpid/cluster/UpdateClient.cpp
    qpid/trunk/qpid/cpp/src/tests/cluster_test.cpp
    qpid/trunk/qpid/cpp/xml/cluster.xml

Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp?rev=893175&r1=893174&r2=893175&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp Tue Dec 22 14:22:17 2009
@@ -269,9 +269,10 @@
     return sessionState().getSemanticState();
 }
 
-void Connection::consumerState(const string& name, bool blocked, bool notifyEnabled)
+void Connection::consumerState(const string& name, bool blocked, bool notifyEnabled,
const SequenceNumber& position)
 {
     broker::SemanticState::ConsumerImpl& c = semanticState().find(name);
+    c.position = position;
     c.setBlocked(blocked);
     if (notifyEnabled) c.enableNotify(); else c.disableNotify();
     consumerNumbering.add(c.shared_from_this());

Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.h?rev=893175&r1=893174&r2=893175&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.h Tue Dec 22 14:22:17 2009
@@ -108,7 +108,7 @@
     // Called for data delivered from the cluster.
     void deliveredFrame(const EventFrame&);
 
-    void consumerState(const std::string& name, bool blocked, bool notifyEnabled);
+    void consumerState(const std::string& name, bool blocked, bool notifyEnabled, const
qpid::framing::SequenceNumber& position);
     
     // ==== Used in catch-up mode to build initial state.
     // 

Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/UpdateClient.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/UpdateClient.cpp?rev=893175&r1=893174&r2=893175&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/UpdateClient.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/UpdateClient.cpp Tue Dec 22 14:22:17 2009
@@ -381,7 +381,8 @@
     ClusterConnectionProxy(shadowSession).consumerState(
         ci->getName(),
         ci->isBlocked(),
-        ci->isNotifyEnabled()
+        ci->isNotifyEnabled(),
+        ci->position
     );
     consumerNumbering.add(ci);
 

Modified: qpid/trunk/qpid/cpp/src/tests/cluster_test.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/cluster_test.cpp?rev=893175&r1=893174&r2=893175&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/cluster_test.cpp (original)
+++ qpid/trunk/qpid/cpp/src/tests/cluster_test.cpp Tue Dec 22 14:22:17 2009
@@ -1149,6 +1149,43 @@
     BOOST_CHECK(!browseByteCredit(c1, "q", size-1, m));
 }
 
+// Test that consumer positions are updated correctly.
+// Regression test for https://bugzilla.redhat.com/show_bug.cgi?id=541927
+//
+QPID_AUTO_TEST_CASE(testUpdateConsumerPosition) {
+    ClusterFixture::Args args;
+    prepareArgs(args, durableFlag);
+    ClusterFixture cluster(1, args, -1);
+    Client c0(cluster[0], "c0");
+
+    c0.session.queueDeclare("q", arg::durable=durableFlag);
+    SubscriptionSettings settings;
+    settings.autoAck = 0;
+    // Set the acquire mode to 'not-acquired' the consumer moves along the queue
+    // but does not acquire (remove) messages.
+    settings.acquireMode = ACQUIRE_MODE_NOT_ACQUIRED;
+    Subscription s = c0.subs.subscribe(c0.lq, "q", settings);
+    c0.session.messageTransfer(arg::content=makeMessage("1", "q", durableFlag));
+    BOOST_CHECK_EQUAL("1", c0.lq.get(TIMEOUT).getData());
+
+    // Add another member, send/receive another message and acquire
+    // the messages.  With the bug, this creates an inconsistency
+    // because the browse position was not updated to the new member.
+    cluster.add();
+    c0.session.messageTransfer(arg::content=makeMessage("2", "q", durableFlag));
+    BOOST_CHECK_EQUAL("2", c0.lq.get(TIMEOUT).getData());
+    s.acquire(s.getUnacquired());
+    s.accept(s.getUnaccepted());
+
+    // In the bug we now have 0 messages on cluster[0] and 1 message on cluster[1]
+    // Subscribing on cluster[1] provokes an error that shuts down cluster[0]
+    Client c1(cluster[1], "c1");
+    Subscription s1 = c1.subs.subscribe(c1.lq, "q"); // Default auto-ack=1
+    Message m;
+    BOOST_CHECK(!c1.lq.get(m, TIMEOUT/10));
+    BOOST_CHECK_EQUAL(c1.session.queueQuery("q").getMessageCount(), 0u);
+    BOOST_CHECK_EQUAL(c0.session.queueQuery("q").getMessageCount(), 0u);
+}
 
 QPID_AUTO_TEST_SUITE_END()
 }} // namespace qpid::tests

Modified: qpid/trunk/qpid/cpp/xml/cluster.xml
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/xml/cluster.xml?rev=893175&r1=893174&r2=893175&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/xml/cluster.xml (original)
+++ qpid/trunk/qpid/cpp/xml/cluster.xml Tue Dec 22 14:22:17 2009
@@ -137,6 +137,7 @@
       <field name="name" type="str8"/>
       <field name="blocked" type="bit"/>
       <field name="notifyEnabled" type="bit"/>
+      <field name="position" type="sequence-no"/>
     </control>
 
     <!-- Delivery-record for outgoing messages sent but not yet accepted. -->



---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project:      http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org


Mime
View raw message