diff options
| author | Robert Gemmell <robbie@apache.org> | 2011-09-07 16:15:44 +0000 |
|---|---|---|
| committer | Robert Gemmell <robbie@apache.org> | 2011-09-07 16:15:44 +0000 |
| commit | 60842f4efee51f9a5766f544aa5d844d3f772ca6 (patch) | |
| tree | 51361e6c1c924df7160e6648a902f431d28d6db4 /java | |
| parent | ee0cdba739ec3d5e4fe5a20dd2804d558e0ccc0d (diff) | |
| download | qpid-python-60842f4efee51f9a5766f544aa5d844d3f772ca6.tar.gz | |
QPID-2901: fixes racing conditions on broker side when TopicDeletePolicy is closing the consumer Session but the broker is trying to deliver next message to the Subscription of this closing Session.
Applied patch from Oleksandr Rudyy <orudyy@gmail.com>.
git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@1166246 13f79535-47bb-0310-9956-ffa450edef68
Diffstat (limited to 'java')
3 files changed, 16 insertions, 2 deletions
diff --git a/java/broker/src/main/java/org/apache/qpid/server/subscription/Subscription_0_10.java b/java/broker/src/main/java/org/apache/qpid/server/subscription/Subscription_0_10.java index 8a5f9ffef5..c5d6bc203c 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/subscription/Subscription_0_10.java +++ b/java/broker/src/main/java/org/apache/qpid/server/subscription/Subscription_0_10.java @@ -194,7 +194,7 @@ public class Subscription_0_10 implements Subscription, FlowCreditManager.FlowCr public boolean isSuspended() { - return !isActive() || _deleted.get(); // TODO check for Session suspension + return !isActive() || _deleted.get() || _session.isClosing(); // TODO check for Session suspension } public boolean hasInterest(QueueEntry entry) diff --git a/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java b/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java index 65790e2e6f..12ef125b2e 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java +++ b/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java @@ -677,4 +677,18 @@ public class ServerSession extends Session implements AuthorizationHolder, Sessi getChannel()) + "] "; } + + @Override + public void close() + { + // unregister subscriptions in order to prevent sending of new messages + // to subscriptions with closing session + final Collection<Subscription_0_10> subscriptions = getSubscriptions(); + for (Subscription_0_10 subscription_0_10 : subscriptions) + { + unregister(subscription_0_10); + } + + super.close(); + } } diff --git a/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java b/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java index c38188cdc3..d9194a3408 100644 --- a/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java +++ b/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java @@ -1183,7 +1183,7 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect if (code != null) { - je = new JMSException(Integer.toString(code.getCode()), "Exception thrown against " + toString() + ": " + cause); + je = new JMSException("Exception thrown against " + toString() + ": " + cause, Integer.toString(code.getCode())); } else { |