QPID-3202: Clustered brokers shut down with "unknown connection" error.

This error is an assertion recently introduced in r1091097, the bug
has probably been there for a while.

Two fixes:
- Connections that close before they are announced no longer send a deliver-close.
- Fixed race in OutputInterceptor that sent a deliver-do-output after deliver-close.

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

1 files changed, 7 insertions, 5 deletions

diff --git a/cpp/src/qpid/cluster/Connection.cpp b/cpp/src/qpid/cluster/Connection.cpp
index c1304c2b75..f2ea466a9b 100644
--- a/cpp/src/qpid/cluster/Connection.cpp
+++ b/cpp/src/qpid/cluster/Connection.cpp
@@ -78,7 +78,7 @@ const std::string shadowPrefix("[shadow]");
 Connection::Connection(Cluster& c, sys::ConnectionOutputHandler& out,
                        const std::string& mgmtId,
                        const ConnectionId& id, const qpid::sys::SecuritySettings& external)
-    : cluster(c), self(id), catchUp(false), output(*this, out),
+    : cluster(c), self(id), catchUp(false), announced(false), output(*this, out),
       connectionCtor(&output, cluster.getBroker(), mgmtId, external, false, 0, true),
       expectProtocolHeader(false),
       mcastFrameHandler(cluster.getMulticast(), self),
@@ -90,7 +90,7 @@ Connection::Connection(Cluster& c, sys::ConnectionOutputHandler& out,
 Connection::Connection(Cluster& c, sys::ConnectionOutputHandler& out,
                        const std::string& mgmtId, MemberId member,
                        bool isCatchUp, bool isLink, const qpid::sys::SecuritySettings& external
-) : cluster(c), self(member, ++idCounter), catchUp(isCatchUp), output(*this, out),
+) : cluster(c), self(member, ++idCounter), catchUp(isCatchUp), announced(false), output(*this, out),
     connectionCtor(&output, cluster.getBroker(),
                    mgmtId,
                    external,
@@ -255,7 +255,7 @@ void Connection::deliveredFrame(const EventFrame& f) {
     }
 }
 
-// A local connection is closed by the network layer.
+// A local connection is closed by the network layer. Called in the connection thread.
 void Connection::closed() {
     try {
         if (isUpdated()) {
@@ -272,8 +272,9 @@ void Connection::closed() {
             // closed and process any outstanding frames from the cluster
             // until self-delivery of deliver-close.
             output.closeOutput();
-            cluster.getMulticast().mcastControl(
-                ClusterConnectionDeliverCloseBody(), self);
+            if (announced)
+                cluster.getMulticast().mcastControl(
+                    ClusterConnectionDeliverCloseBody(), self);
         }
     }
     catch (const std::exception& e) {
@@ -384,6 +385,7 @@ void Connection::processInitialFrames(const char*& ptr, size_t size) {
                 connection->getUserId(),
                 initialFrames),
             getId());
+        announced = true;
         initialFrames.clear();
     }
 }