QPID-4680 : [Java Broker] NoAck subscription on persistent queue in 0-8/9/9-1 codepath leaves queue entry records in database

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

1 files changed, 27 insertions, 5 deletions

diff --git a/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionImpl.java b/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionImpl.java
index 6c5cb2e721..a8acca51e9 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionImpl.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionImpl.java
@@ -42,6 +42,8 @@ import org.apache.qpid.server.protocol.AMQProtocolSession;
 import org.apache.qpid.server.protocol.AMQSessionModel;
 import org.apache.qpid.server.queue.AMQQueue;
 import org.apache.qpid.server.queue.QueueEntry;
+import org.apache.qpid.server.txn.AutoCommitTransaction;
+import org.apache.qpid.server.txn.ServerTransaction;
 
 import java.util.Map;
 import java.util.UUID;
@@ -92,7 +94,7 @@ public abstract class SubscriptionImpl implements Subscription, FlowCreditManage
     private final AtomicLong _unacknowledgedBytes = new AtomicLong(0);
 
     private long _createTime = System.currentTimeMillis();
-    
+
 
     static final class BrowserSubscription extends SubscriptionImpl
     {
@@ -146,6 +148,8 @@ public abstract class SubscriptionImpl implements Subscription, FlowCreditManage
 
     public static class NoAckSubscription extends SubscriptionImpl
     {
+        private volatile AutoCommitTransaction _txn;
+
         public NoAckSubscription(AMQChannel channel, AMQProtocolSession protocolSession,
                                  AMQShortString consumerTag, FieldTable filters,
                                  boolean noLocal, FlowCreditManager creditManager,
@@ -190,8 +194,13 @@ public abstract class SubscriptionImpl implements Subscription, FlowCreditManage
 
             // The send may of course still fail, in which case, as
             // the message is unacked, it will be lost.
-            entry.dequeue();
+            if(_txn == null)
+            {
+                _txn = new AutoCommitTransaction(getQueue().getVirtualHost().getMessageStore());
+            }
+            _txn.dequeue(getQueue(), entry.getMessage(), NOOP);
 
+            entry.dequeue();
 
             synchronized (getChannel())
             {
@@ -212,6 +221,19 @@ public abstract class SubscriptionImpl implements Subscription, FlowCreditManage
             return false;
         }
 
+        private static final ServerTransaction.Action NOOP =
+                new ServerTransaction.Action()
+                {
+                    @Override
+                    public void postCommit()
+                    {
+                    }
+
+                    @Override
+                    public void onRollback()
+                    {
+                    }
+                };
     }
 
     /**
@@ -275,7 +297,7 @@ public abstract class SubscriptionImpl implements Subscription, FlowCreditManage
         public void send(QueueEntry entry, boolean batch) throws AMQException
         {
 
-            
+
             synchronized (getChannel())
             {
                 getChannel().getProtocolSession().setDeferFlush(batch);
@@ -576,7 +598,7 @@ public abstract class SubscriptionImpl implements Subscription, FlowCreditManage
     {
         return _consumerTag == null ? null : _consumerTag.asString();
     }
-    
+
     public long getSubscriptionID()
     {
         return _subscriptionID;
@@ -821,7 +843,7 @@ public abstract class SubscriptionImpl implements Subscription, FlowCreditManage
             }
         });
     }
-    
+
     public long getUnacknowledgedBytes()
     {
         return _unacknowledgedBytes.longValue();