From c4f018f7c10d2169ced4c59e776844ee5cf52d33 Mon Sep 17 00:00:00 2001 From: Martin Ritchie Date: Tue, 13 Mar 2007 10:35:42 +0000 Subject: QPID-346,QPID-386,QPID-403, QPID-410 Rollback, Basic-Reject, QueueBrowser NO_ACK. QPID-346 Message loss after rollback\recover QPID-386 Updated Transactional Tests to cover underlying AMQP/Qpid state. QPID-403 Implement Basic.Reject QPID-410 Queue Browsers should use not acknowledge messages. ------------------------------------- Broker TxAck - Added comment and fixed white space UnacknowledgedMessage - Added comment for messageDecrement AMQChannel - Added extra debugging. + Created a NonTransactionalContext for requeuing messages as using txContext will tie the requeue to any runing transaction. + Updated message reference counting. So it is in terms of queues don't increment when giving to client. BasicCancelMethodHandler - Added Debug log. BasicConsumeMethodHandler - Reverted to directly writes frames to the session, throwing ChannelException caused problems. Added Trace and debug logging. BasicRejectMethodHandler, ChannelCloseHandler, ConnectionCloseMethodHandler - Added Debug logging AMQPFastProtocolHandler - moved error log to before session.write AMQMessage - Added additional debug via debugIdentity() and comments AMQQueue - Decoupled reference counting from dequeue operation. ConcurrentSelectorDeliveryManager - Added comments and increased info in debug logging SubscriptionImpl - Disabled use of acks for browsers. For now put setDeliveredToConsumer back in the finally block. commented that I'm not sure this is correct as even an error writing to client will cause msg to be marked delivered to consumer. + On Close ensured that it is only called once. + Had problem where closing browser was causing two CancelOk frames to be sent back to client. RequiredDeliveryException - Added comment to explain incrementReference LocalTransactionalContext - Commented out incrementReference as it shouldn't be required here. NonTransactionalContext - Removed incrementReference on deliver + - Fixed bug where browsers - acks would cause messages to be discarded. new JIRA this needs tidied up. TxnBuffer - Added debug logging. Client ------ AMQQueueBrowser - Added comments AMQSession - Added comments and debug + Updated to cause closed consumer to reject messages rather than receive them. + Prevented NoConsumer's from rollingback and rejecting.. they simply clear their SyncQueue - JIRA to ensure clean state with rollback BasicMessageConsumer - Added trace level debuging on close calls + Forced noConsume-rs to use NO_ACK + added more logging Closeable - Updated to use isClosed rather than directly calling _closed.get() to aid in future work on ensuring multi threaded close still allows pending acks to be processed first. ChannelCloseOkMethodHandler - updated comment AMQProtocolSession - Update comments,whitespace TransportConnection - removed static block FlowControllingBlockingQueue - Added isEmpty() Method PropertyValueTest - Added VM Broker setup + Updated test to run once and 50 times to pull out delivery tag problems that were occuring. + Adjusted logging level to be more helpful. moved some info down to trace and debug. MessageRequeueTest - Moved QpidClientConnection its own file. + Fixed it so it actually runs more than one consumer, concurrently.Now 3 was 1. ConcurrentLinkedMessageQueueAtomicSize - Implemented iterator(). Added QueueBrowserTest to system tests to test QueueBrowsering. git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@517638 13f79535-47bb-0310-9956-ffa450edef68 --- .../java/org/apache/qpid/server/AMQChannel.java | 114 +++++++++++++++------ .../qpid/server/RequiredDeliveryException.java | 4 + .../java/org/apache/qpid/server/ack/TxAck.java | 4 +- .../qpid/server/ack/UnacknowledgedMessage.java | 2 +- .../server/handler/BasicCancelMethodHandler.java | 9 ++ .../server/handler/BasicConsumeMethodHandler.java | 42 +++++++- .../server/handler/BasicRejectMethodHandler.java | 18 ++-- .../qpid/server/handler/ChannelCloseHandler.java | 7 +- .../handler/ConnectionCloseMethodHandler.java | 11 +- .../server/protocol/AMQPFastProtocolHandler.java | 3 +- .../org/apache/qpid/server/queue/AMQMessage.java | 30 +++--- .../org/apache/qpid/server/queue/AMQQueue.java | 3 +- .../queue/ConcurrentSelectorDeliveryManager.java | 12 ++- .../apache/qpid/server/queue/SubscriptionImpl.java | 109 +++++++++++++------- .../qpid/server/txn/LocalTransactionalContext.java | 2 +- .../qpid/server/txn/NonTransactionalContext.java | 22 +++- .../java/org/apache/qpid/server/txn/TxnBuffer.java | 12 ++- 17 files changed, 290 insertions(+), 114 deletions(-) (limited to 'java/broker/src') diff --git a/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java b/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java index 5dd6619cff..1ebe5fa0a2 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java +++ b/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java @@ -319,6 +319,25 @@ public class AMQChannel public void unsubscribeConsumer(AMQProtocolSession session, AMQShortString consumerTag) throws AMQException { + if (_log.isDebugEnabled()) + { + _log.debug("Unacked Map Dump size:" + _unacknowledgedMessageMap.size()); + _unacknowledgedMessageMap.visit(new UnacknowledgedMessageMap.Visitor() + { + + public boolean callback(UnacknowledgedMessage message) throws AMQException + { + _log.debug(message); + + return true; + } + + public void visitComplete() + { + } + }); + } + AMQQueue q = _consumerTag2QueueMap.remove(consumerTag); if (q != null) { @@ -342,9 +361,23 @@ public class AMQChannel private void unsubscribeAllConsumers(AMQProtocolSession session) throws AMQException { - _log.info("Unsubscribing all consumers on channel " + toString()); + if (_log.isInfoEnabled()) + { + if (!_consumerTag2QueueMap.isEmpty()) + { + _log.info("Unsubscribing all consumers on channel " + toString()); + } + else + { + _log.info("No consumers to unsubscribe on channel " + toString()); + } + } for (Map.Entry me : _consumerTag2QueueMap.entrySet()) { + if (_log.isInfoEnabled()) + { + _log.info("Unsubscribing consumer '" + me.getKey() + "' on channel " + toString()); + } me.getValue().unregisterProtocolSession(session, _channelId, me.getKey()); } _consumerTag2QueueMap.clear(); @@ -369,7 +402,11 @@ public class AMQChannel } else { - _log.debug(debugIdentity() + " Adding unacked message(" + deliveryTag + ") with a queue(" + queue + "):" + message.debugIdentity()); + if (_log.isDebugEnabled()) + { + _log.debug(debugIdentity() + " Adding unacked message(" + message.toString() + " DT:" + deliveryTag + + ") with a queue(" + queue + ") for " + consumerTag); + } } } @@ -395,25 +432,38 @@ public class AMQChannel */ public void requeue() throws AMQException { + if (_log.isInfoEnabled()) + { + _log.info("Requeuing for " + toString()); + } + // we must create a new map since all the messages will get a new delivery tag when they are redelivered Collection messagesToBeDelivered = _unacknowledgedMessageMap.cancelAllMessages(); + if (_log.isDebugEnabled()) + { + _log.info("Requeuing " + messagesToBeDelivered.size() + " unacked messages."); + } // Deliver these messages out of the transaction as their delivery was never // part of the transaction only the receive. - TransactionalContext deliveryContext; - if (!(_txnContext instanceof NonTransactionalContext)) + TransactionalContext deliveryContext = null; + + if (!messagesToBeDelivered.isEmpty()) { - if (_nonTransactedContext == null) + if (!(_txnContext instanceof NonTransactionalContext)) { - _nonTransactedContext = new NonTransactionalContext(_messageStore, _storeContext, this, - _returnMessages, _browsedAcks); - } +// if (_nonTransactedContext == null) + { + _nonTransactedContext = new NonTransactionalContext(_messageStore, _storeContext, this, + _returnMessages, _browsedAcks); + } - deliveryContext = _nonTransactedContext; - } - else - { - deliveryContext = _txnContext; + deliveryContext = _nonTransactedContext; + } + else + { + deliveryContext = _txnContext; + } } @@ -421,6 +471,10 @@ public class AMQChannel { if (unacked.queue != null) { + // Ensure message is released for redelivery + unacked.message.release(); + + // Mark message redelivered unacked.message.setRedelivered(true); // Deliver Message @@ -459,7 +513,7 @@ public class AMQChannel TransactionalContext deliveryContext; if (!(_txnContext instanceof NonTransactionalContext)) { - if (_nonTransactedContext == null) +// if (_nonTransactedContext == null) { _nonTransactedContext = new NonTransactionalContext(_messageStore, _storeContext, this, _returnMessages, _browsedAcks); @@ -472,13 +526,12 @@ public class AMQChannel deliveryContext = _txnContext; } - if (unacked.queue != null) { //Redeliver the messages to the front of the queue deliveryContext.deliver(unacked.message, unacked.queue, true); - - unacked.message.decrementReference(_storeContext); + //Deliver increments the message count but we have already deliverted this once so don't increment it again + // this was because deliver did an increment changed this. } else { @@ -489,7 +542,6 @@ public class AMQChannel // // deliveryContext.deliver(unacked.message, getDeadLetterQueue(), false); // -// unacked.message.decrementReference(_storeContext); } } else @@ -656,15 +708,16 @@ public class AMQChannel } sub.addToResendQueue(msg); _unacknowledgedMessageMap.remove(message.deliveryTag); - // Don't decrement as we are bypassing the normal deliver which increments - // this is why there is a decrement on the Requeue as deliver will increment. - // msg.decrementReference(_storeContext); } } // sync(sub.getSendLock) } else { - _log.info("DeliveredSubscription not recorded so just requeueing to prevent loss"); + + if (_log.isInfoEnabled()) + { + _log.info("DeliveredSubscription not recorded so just requeueing(" + message.toString() + ")to prevent loss"); + } //move this message to requeue msgToRequeue.add(message); } @@ -706,7 +759,6 @@ public class AMQChannel deliveryContext.deliver(message.message, message.queue, true); _unacknowledgedMessageMap.remove(message.deliveryTag); - message.message.decrementReference(_storeContext); } } @@ -760,8 +812,18 @@ public class AMQChannel { synchronized (_unacknowledgedMessageMap.getLock()) { + if (_log.isDebugEnabled()) + { + _log.debug("Unacked (PreAck) Size:" + _unacknowledgedMessageMap.size()); + } + _unacknowledgedMessageMap.acknowledgeMessage(deliveryTag, multiple, _txnContext); checkSuspension(); + if (_log.isDebugEnabled()) + { + _log.debug("Unacked (PostAck) Size:" + _unacknowledgedMessageMap.size()); + } + } } @@ -775,12 +837,6 @@ public class AMQChannel return _unacknowledgedMessageMap; } - public void addUnacknowledgedBrowsedMessage(AMQMessage msg, long deliveryTag, AMQShortString consumerTag, AMQQueue queue) - { - _browsedAcks.add(deliveryTag); - addUnacknowledgedMessage(msg, deliveryTag, consumerTag, queue); - } - private void checkSuspension() { boolean suspend; diff --git a/java/broker/src/main/java/org/apache/qpid/server/RequiredDeliveryException.java b/java/broker/src/main/java/org/apache/qpid/server/RequiredDeliveryException.java index 820f0122f5..fb16267d97 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/RequiredDeliveryException.java +++ b/java/broker/src/main/java/org/apache/qpid/server/RequiredDeliveryException.java @@ -37,6 +37,10 @@ public abstract class RequiredDeliveryException extends AMQException { super(message); _amqMessage = payload; + // Increment the reference as this message is in the routing phase + // and so will have the ref decremented as routing fails. + // we need to keep this message around so we can return it in the + // handler. So increment here. payload.incrementReference(); } diff --git a/java/broker/src/main/java/org/apache/qpid/server/ack/TxAck.java b/java/broker/src/main/java/org/apache/qpid/server/ack/TxAck.java index c987c12154..aac9408247 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/ack/TxAck.java +++ b/java/broker/src/main/java/org/apache/qpid/server/ack/TxAck.java @@ -101,6 +101,8 @@ public class TxAck implements TxnOp for (UnacknowledgedMessage msg : _unacked) { msg.restoreTransientMessageData(); + + //Message has been ack so discard it. This will dequeue and decrement the reference. msg.discard(storeContext); } } @@ -124,7 +126,7 @@ public class TxAck implements TxnOp _map.remove(_unacked); for (UnacknowledgedMessage msg : _unacked) { - msg.clearTransientMessageData(); + msg.clearTransientMessageData(); } } diff --git a/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessage.java b/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessage.java index 940b5b2bf1..b8c5e821f7 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessage.java +++ b/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessage.java @@ -39,7 +39,6 @@ public class UnacknowledgedMessage this.message = message; this.consumerTag = consumerTag; this.deliveryTag = deliveryTag; - message.incrementReference(); } public String toString() @@ -63,6 +62,7 @@ public class UnacknowledgedMessage { message.dequeue(storeContext, queue); } + //if the queue is null then the message is waiting to be acked, but has been removed. message.decrementReference(storeContext); } diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/BasicCancelMethodHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/BasicCancelMethodHandler.java index 7d18043f5c..8bab96a11b 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/handler/BasicCancelMethodHandler.java +++ b/java/broker/src/main/java/org/apache/qpid/server/handler/BasicCancelMethodHandler.java @@ -29,9 +29,12 @@ import org.apache.qpid.server.AMQChannel; import org.apache.qpid.server.protocol.AMQProtocolSession; import org.apache.qpid.server.state.AMQStateManager; import org.apache.qpid.server.state.StateAwareMethodListener; +import org.apache.log4j.Logger; public class BasicCancelMethodHandler implements StateAwareMethodListener { + private static final Logger _log = Logger.getLogger(BasicCancelMethodHandler.class); + private static final BasicCancelMethodHandler _instance = new BasicCancelMethodHandler(); public static BasicCancelMethodHandler getInstance() @@ -55,6 +58,12 @@ public class BasicCancelMethodHandler implements StateAwareMethodListener +public class ConnectionCloseMethodHandler implements StateAwareMethodListener { private static final Logger _logger = Logger.getLogger(ConnectionCloseMethodHandler.class); @@ -49,8 +49,11 @@ public class ConnectionCloseMethodHandler implements StateAwareMethodListener destinationQueues = _transientMessageData.getDestinationQueues(); if (_log.isDebugEnabled()) { - _log.debug("Delivering message " + _messageId + " to " + destinationQueues); + _log.debug("Delivering message " + debugIdentity() + " to " + destinationQueues); } try { @@ -589,6 +590,8 @@ public class AMQMessage for (AMQQueue q : destinationQueues) { + //Increment the references to this message for each queue delivery. + incrementReference(); //normal deliver so add this message at the end. _txnContext.deliver(this, q, false); } @@ -596,6 +599,7 @@ public class AMQMessage finally { destinationQueues.clear(); + // Remove refence for routing process . Reference count should now == delivered queue count decrementReference(storeContext); } } diff --git a/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java b/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java index 7c2fe73386..78f144703b 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java +++ b/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java @@ -516,7 +516,7 @@ public class AMQQueue implements Managable, Comparable { if (_logger.isInfoEnabled()) { - _logger.warn("Auto-deleteing queue:" + this); + _logger.info("Auto-deleteing queue:" + this); } autodelete(); // we need to manually fire the event to the removed subscription (which was the last one left for this @@ -624,7 +624,6 @@ public class AMQQueue implements Managable, Comparable try { msg.dequeue(storeContext, this); - msg.decrementReference(storeContext); } catch (MessageCleanupException e) { diff --git a/java/broker/src/main/java/org/apache/qpid/server/queue/ConcurrentSelectorDeliveryManager.java b/java/broker/src/main/java/org/apache/qpid/server/queue/ConcurrentSelectorDeliveryManager.java index 87868f0b25..6122d191f8 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/queue/ConcurrentSelectorDeliveryManager.java +++ b/java/broker/src/main/java/org/apache/qpid/server/queue/ConcurrentSelectorDeliveryManager.java @@ -383,6 +383,9 @@ public class ConcurrentSelectorDeliveryManager implements DeliveryManager return count; } + /** + This can only be used to clear the _messages queue. Any subscriber resend queue will not be purged. + */ private AMQMessage getNextMessage() throws AMQException { return getNextMessage(_messages, null); @@ -392,13 +395,14 @@ public class ConcurrentSelectorDeliveryManager implements DeliveryManager { AMQMessage message = messages.peek(); - while (message != null && ((sub != null && sub.isBrowser()) || message.taken(sub))) + //while (we have a message) && (The subscriber is not a browser or we are clearing) && (Check message is taken.) + while (message != null && (sub != null && !sub.isBrowser() || sub == null) && message.taken(sub)) { //remove the already taken message AMQMessage removed = messages.poll(); assert removed == message; - + _totalMessageSize.addAndGet(-message.getSize()); if (_log.isTraceEnabled()) @@ -494,7 +498,7 @@ public class ConcurrentSelectorDeliveryManager implements DeliveryManager _extraMessages.decrementAndGet(); } - else if (messageQueue == sub.getPreDeliveryQueue()) + else if (messageQueue == sub.getPreDeliveryQueue() && !sub.isBrowser()) { if (_log.isInfoEnabled()) { @@ -695,7 +699,7 @@ public class ConcurrentSelectorDeliveryManager implements DeliveryManager { if (_log.isDebugEnabled()) { - _log.debug(debugIdentity() + " Message(" + msg.debugIdentity() + + _log.debug(debugIdentity() + " Message(" + msg.toString() + ") has been taken so disregarding deliver request to Subscriber:" + System.identityHashCode(s)); } diff --git a/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionImpl.java b/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionImpl.java index 20033daac7..d3578d39e8 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionImpl.java +++ b/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionImpl.java @@ -256,10 +256,10 @@ public class SubscriptionImpl implements Subscription // We don't need to add the message to the unacknowledgedMap as we don't need to know if the client // received the message. If it is lost in transit that is not important. - if (_acks) - { - channel.addUnacknowledgedBrowsedMessage(msg, deliveryTag, consumerTag, queue); - } +// if (_acks) +// { +// channel.addUnacknowledgedBrowsedMessage(msg, deliveryTag, consumerTag, queue); +// } if (_sendLock.get()) { @@ -273,41 +273,49 @@ public class SubscriptionImpl implements Subscription private void sendToConsumer(StoreContext storeContext, AMQMessage msg, AMQQueue queue) throws AMQException { - // if we do not need to wait for client acknowledgements - // we can decrement the reference count immediately. + try + { // if we do not need to wait for client acknowledgements + // we can decrement the reference count immediately. - // By doing this _before_ the send we ensure that it - // doesn't get sent if it can't be dequeued, preventing - // duplicate delivery on recovery. + // By doing this _before_ the send we ensure that it + // doesn't get sent if it can't be dequeued, preventing + // duplicate delivery on recovery. - // The send may of course still fail, in which case, as - // the message is unacked, it will be lost. - if (!_acks) - { - if (_logger.isDebugEnabled()) + // The send may of course still fail, in which case, as + // the message is unacked, it will be lost. + if (!_acks) { - _logger.debug("No ack mode so dequeuing message immediately: " + msg.getMessageId()); + if (_logger.isDebugEnabled()) + { + _logger.debug("No ack mode so dequeuing message immediately: " + msg.getMessageId()); + } + queue.dequeue(storeContext, msg); } - queue.dequeue(storeContext, msg); - } - synchronized (channel) - { - long deliveryTag = channel.getNextDeliveryTag(); - if (_sendLock.get()) + synchronized (channel) { - _logger.error("Sending " + msg + " when subscriber(" + this + ") is closed!"); - } + long deliveryTag = channel.getNextDeliveryTag(); - if (_acks) - { - channel.addUnacknowledgedMessage(msg, deliveryTag, consumerTag, queue); - msg.decrementReference(storeContext); - } + if (_sendLock.get()) + { + _logger.error("Sending " + msg + " when subscriber(" + this + ") is closed!"); + } - protocolSession.getProtocolOutputConverter().writeDeliver(msg, channel.getChannelId(), deliveryTag, consumerTag); + if (_acks) + { + channel.addUnacknowledgedMessage(msg, deliveryTag, consumerTag, queue); + } + + protocolSession.getProtocolOutputConverter().writeDeliver(msg, channel.getChannelId(), deliveryTag, consumerTag); + + } + } + finally + { //Only set delivered if it actually was writen successfully.. - // using a try->finally would set it even if an error occured. + // using a try->finally would set it even if an error occured. + // Is this what we want? + msg.setDeliveredToConsumer(); } } @@ -461,14 +469,25 @@ public class SubscriptionImpl implements Subscription public void close() { + boolean closed = false; synchronized (_sendLock) { if (_logger.isDebugEnabled()) { - _logger.debug("Setting SendLock true"); + _logger.debug("Setting SendLock true:" + debugIdentity()); + } + + closed = _sendLock.getAndSet(true); + } + + if (closed) + { + if (_logger.isDebugEnabled()) + { + _logger.debug("Called close() on a closed subscription"); } - _sendLock.set(true); + return; } if (_logger.isInfoEnabled()) @@ -488,16 +507,36 @@ public class SubscriptionImpl implements Subscription //remove references in PDQ if (_messages != null) { + if (_logger.isInfoEnabled()) + { + _logger.info("Clearing PDQ (" + debugIdentity() + "):" + this); + } + _messages.clear(); } + } + + private void autoclose() + { + close(); if (_autoClose && !_sentClose) { - _logger.info("Closing autoclose subscription:" + this); + _logger.info("Closing autoclose subscription (" + debugIdentity() + "):" + this); + ProtocolOutputConverter converter = protocolSession.getProtocolOutputConverter(); converter.confirmConsumerAutoClose(channel.getChannelId(), consumerTag); - _sentClose = true; + + //fixme JIRA do this better + try + { + channel.unsubscribeConsumer(protocolSession, consumerTag); + } + catch (AMQException e) + { + // Occurs if we cannot find the subscriber in the channel with protocolSession and consumerTag. + } } } @@ -590,7 +629,7 @@ public class SubscriptionImpl implements Subscription { if (_messages.isEmpty()) { - close(); + autoclose(); return null; } } diff --git a/java/broker/src/main/java/org/apache/qpid/server/txn/LocalTransactionalContext.java b/java/broker/src/main/java/org/apache/qpid/server/txn/LocalTransactionalContext.java index e5cce672f6..cf0da55f2a 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/txn/LocalTransactionalContext.java +++ b/java/broker/src/main/java/org/apache/qpid/server/txn/LocalTransactionalContext.java @@ -100,7 +100,7 @@ public class LocalTransactionalContext implements TransactionalContext // be added for every queue onto which the message is // enqueued. Finally a cleanup op will be added to decrement // the reference associated with the routing. - message.incrementReference(); +// message.incrementReference(); _postCommitDeliveryList.add(new DeliveryDetails(message, queue, deliverFirst)); _messageDelivered = true; /*_txnBuffer.enlist(new DeliverMessageOperation(message, queue)); diff --git a/java/broker/src/main/java/org/apache/qpid/server/txn/NonTransactionalContext.java b/java/broker/src/main/java/org/apache/qpid/server/txn/NonTransactionalContext.java index 19146da22e..181dfa3a80 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/txn/NonTransactionalContext.java +++ b/java/broker/src/main/java/org/apache/qpid/server/txn/NonTransactionalContext.java @@ -93,7 +93,6 @@ public class NonTransactionalContext implements TransactionalContext { try { - message.incrementReference(); queue.process(_storeContext, message, deliverFirst); //following check implements the functionality //required by the 'immediate' flag: @@ -128,6 +127,8 @@ public class NonTransactionalContext implements TransactionalContext { _log.debug("Discarding message: " + message.message.getMessageId()); } + + //Message has been ack so discard it. This will dequeue and decrement the reference. message.discard(_storeContext); } else @@ -160,6 +161,8 @@ public class NonTransactionalContext implements TransactionalContext { _log.debug("Discarding message: " + msg.message.getMessageId()); } + + //Message has been ack so discard it. This will dequeue and decrement the reference. msg.discard(_storeContext); } else @@ -181,7 +184,22 @@ public class NonTransactionalContext implements TransactionalContext throw new AMQException("Single ack on delivery tag " + deliveryTag + " not known for channel:" + _channel.getChannelId()); } - msg.discard(_storeContext); + + if (!_browsedAcks.contains(deliveryTag)) + { + if (_log.isDebugEnabled()) + { + _log.debug("Discarding message: " + msg.message.getMessageId()); + } + + //Message has been ack so discard it. This will dequeue and decrement the reference. + msg.discard(_storeContext); + } + else + { + _browsedAcks.remove(deliveryTag); + } + if (_log.isDebugEnabled()) { _log.debug("Received non-multiple ack for messaging with delivery tag " + deliveryTag + " msg id " + diff --git a/java/broker/src/main/java/org/apache/qpid/server/txn/TxnBuffer.java b/java/broker/src/main/java/org/apache/qpid/server/txn/TxnBuffer.java index d04b93a469..339ca8ae1a 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/txn/TxnBuffer.java +++ b/java/broker/src/main/java/org/apache/qpid/server/txn/TxnBuffer.java @@ -27,10 +27,7 @@ import org.apache.log4j.Logger; import org.apache.qpid.AMQException; import org.apache.qpid.server.store.StoreContext; -/** - * Holds a list of TxnOp instance representing transactional - * operations. - */ +/** Holds a list of TxnOp instance representing transactional operations. */ public class TxnBuffer { private final List _ops = new ArrayList(); @@ -42,6 +39,11 @@ public class TxnBuffer public void commit(StoreContext context) throws AMQException { + if (_log.isDebugEnabled()) + { + _log.debug("Committing " + _ops.size() + " ops to commit.:" + _ops.toArray()); + } + if (prepare(context)) { for (TxnOp op : _ops) @@ -64,7 +66,7 @@ public class TxnBuffer catch (Exception e) { //compensate previously prepared ops - for(int j = 0; j < i; j++) + for (int j = 0; j < i; j++) { _ops.get(j).undoPrepare(); } -- cgit v1.2.1