QPID-2956: cluster broker exits with "error deliveryRecord no update message."

The following sequence of events was causing a broker joining the cluster to shutdown:
- a client acquires or browses a message with TTL set.
- the message expires.
- a new broker joins before the client has acknowledged the message.

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

1 files changed, 11 insertions, 6 deletions

diff --git a/cpp/src/qpid/cluster/ExpiryPolicy.cpp b/cpp/src/qpid/cluster/ExpiryPolicy.cpp
index d91437c5ba..d9a7b0122a 100644
--- a/cpp/src/qpid/cluster/ExpiryPolicy.cpp
+++ b/cpp/src/qpid/cluster/ExpiryPolicy.cpp
@@ -31,7 +31,7 @@ namespace qpid {
 namespace cluster {
 
 ExpiryPolicy::ExpiryPolicy(Multicaster& m, const MemberId& id, sys::Timer& t)
-    : expiryId(0), expiredPolicy(new Expired), mcast(m), memberId(id), timer(t) {}
+    : expiryId(1), expiredPolicy(new Expired), mcast(m), memberId(id), timer(t) {}
 
 struct ExpiryTask : public sys::TimerTask {
     ExpiryTask(const boost::intrusive_ptr<ExpiryPolicy>& policy, uint64_t id, sys::AbsTime when)
@@ -61,12 +61,17 @@ void ExpiryPolicy::willExpire(broker::Message& m) {
         // them as independenty messages so we can have multiple messages
         // with the same expiry ID.
         //
-        // TODO: fix update to avoid duplicating messages.
         sys::Mutex::ScopedLock l(lock);
-        id = expiryId++;        // if this is an update, this expiryId may already exist
-        assert(unexpiredByMessage.find(&m) == unexpiredByMessage.end());
-        unexpiredById.insert(IdMessageMap::value_type(id, &m));
-        unexpiredByMessage[&m] = id;
+        id = expiryId++;
+        if (!id) {              // This is an update of an already-expired message.
+            m.setExpiryPolicy(expiredPolicy);
+        }
+        else {
+            assert(unexpiredByMessage.find(&m) == unexpiredByMessage.end());
+            // If this is an update, the id may already exist
+            unexpiredById.insert(IdMessageMap::value_type(id, &m));
+            unexpiredByMessage[&m] = id;
+        }
     }
     timer.add(new ExpiryTask(this, id, m.getExpiration()));
 }