QPID-3526, QPID-3524: make sure the 0-10 client message.acknowledge() actually acknowledges messages immediately, and does so synchronously, adding test to verify behaviour. Split acknowledge() and commit() methods into version specific session implementations for clarity/reuse, align 0-10 and 0-8/9 transacted publishing behaviour, refactor preDeliver and postDeliver methods, remove dead code from consumers.

Applied patch from Oleksandr Rudyy<orudyy@gmail.com> and myself.

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

14 files changed, 205 insertions, 574 deletions

diff --git a/java/client/src/main/java/org/apache/qpid/client/AMQSession.java b/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
index 566c0320e9..a41f2f9b17 100644
--- a/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
+++ b/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
@@ -612,30 +612,27 @@ public abstract class AMQSession<C extends BasicMessageConsumer, P extends Basic
         {
             throw new IllegalStateException("Session is already closed");
         }
-        else if (hasFailedOver())
+        else if (hasFailedOverDirty())
         {
+            //perform an implicit recover in this scenario
+            recover();
+
+            //notify the consumer
             throw new IllegalStateException("has failed over");
         }
 
-        while (true)
+        try
         {
-            Long tag = _unacknowledgedMessageTags.poll();
-            if (tag == null)
-            {
-                break;
-            }
-
-            try
-            {
-                acknowledgeMessage(tag, false);
-            }
-            catch (TransportException e)
-            {
-                throw toJMSException("Exception while acknowledging message(s):" + e.getMessage(), e);
-            }
+            acknowledgeImpl();
+        }
+        catch (TransportException e)
+        {
+            throw toJMSException("Exception while acknowledging message(s):" + e.getMessage(), e);
         }
     }
 
+    protected abstract void acknowledgeImpl() throws JMSException;
+
     /**
      * Acknowledge one or many messages.
      *
@@ -844,42 +841,28 @@ public abstract class AMQSession<C extends BasicMessageConsumer, P extends Basic
      * @throws JMSException If the JMS provider fails to commit the transaction due to some internal error. This does
      *                      not mean that the commit is known to have failed, merely that it is not known whether it
      *                      failed or not.
-     * @todo Be aware of possible changes to parameter order as versions change.
      */
     public void commit() throws JMSException
     {
         checkTransacted();
 
-        try
+        //Check that we are clean to commit.
+        if (_failedOverDirty)
         {
-            //Check that we are clean to commit.
-            if (_failedOverDirty)
-            {
-                rollback();
-
-                throw new TransactionRolledBackException("Connection failover has occured since last send. " +
-                                                         "Forced rollback");
-            }
+            rollback();
 
+            throw new TransactionRolledBackException("Connection failover has occured with uncommitted transaction activity." +
+                                                     "The session transaction was rolled back.");
+        }
 
-            // Acknowledge all delivered messages
-            while (true)
-            {
-                Long tag = _deliveredMessageTags.poll();
-                if (tag == null)
-                {
-                    break;
-                }
-
-                acknowledgeMessage(tag, false);
-            }
-            // Commits outstanding messages and acknowledgments
-            sendCommit();
+        try
+        {
+            commitImpl();
             markClean();
         }
         catch (AMQException e)
         {
-            throw new JMSAMQException("Failed to commit: " + e.getMessage() + ":" + e.getCause(), e);
+            throw new JMSAMQException("Exception during commit: " + e.getMessage() + ":" + e.getCause(), e);
         }
         catch (FailoverException e)
         {
@@ -891,8 +874,7 @@ public abstract class AMQSession<C extends BasicMessageConsumer, P extends Basic
         }
     }
 
-    public abstract void sendCommit() throws AMQException, FailoverException;
-
+    protected abstract void commitImpl() throws AMQException, FailoverException, TransportException;
 
     public void confirmConsumerCancelled(int consumerTag)
     {
@@ -1580,10 +1562,8 @@ public abstract class AMQSession<C extends BasicMessageConsumer, P extends Basic
 
     abstract public void sync() throws AMQException;
 
-    public int getAcknowledgeMode() throws JMSException
+    public int getAcknowledgeMode()
     {
-        checkNotClosed();
-
         return _acknowledgeMode;
     }
 
@@ -1643,10 +1623,8 @@ public abstract class AMQSession<C extends BasicMessageConsumer, P extends Basic
         return _ticket;
     }
 
-    public boolean getTransacted() throws JMSException
+    public boolean getTransacted()
     {
-        checkNotClosed();
-
         return _transacted;
     }
 
@@ -3096,21 +3074,11 @@ public abstract class AMQSession<C extends BasicMessageConsumer, P extends Basic
      *
      * @return boolean true if failover has occured.
      */
-    public boolean hasFailedOver()
+    public boolean hasFailedOverDirty()
     {
         return _failedOverDirty;
     }
 
-    /**
-     * Check to see if any message have been sent in this transaction and have not been commited.
-     *
-     * @return boolean true if a message has been sent but not commited
-     */
-    public boolean isDirty()
-    {
-        return _dirty;
-    }
-
     public void setTicket(int ticket)
     {
         _ticket = ticket;
diff --git a/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java b/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java
index 0cca946f19..8f77cc6258 100644
--- a/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java
+++ b/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java
@@ -412,25 +412,6 @@ public class AMQSession_0_10 extends AMQSession<BasicMessageConsumer_0_10, Basic
         }
     }
 
-
-    /**
-     * Commit the receipt and the delivery of all messages exchanged by this session resources.
-     */
-    public void sendCommit() throws AMQException, FailoverException
-    {
-        getQpidSession().setAutoSync(true);
-        try
-        {
-            getQpidSession().txCommit();
-        }
-        finally
-        {
-            getQpidSession().setAutoSync(false);
-        }
-        // We need to sync so that we get notify of an error.
-        sync();
-    }
-
     /**
      * Create a queue with a given name.
      *
@@ -463,6 +444,14 @@ public class AMQSession_0_10 extends AMQSession<BasicMessageConsumer_0_10, Basic
     public void sendRecover() throws AMQException, FailoverException
     {
         // release all unacked messages
+        RangeSet ranges = gatherUnackedRangeSet();
+        getQpidSession().messageRelease(ranges, Option.SET_REDELIVERED);
+        // We need to sync so that we get notify of an error.
+        sync();
+    }
+
+    private RangeSet gatherUnackedRangeSet()
+    {
         RangeSet ranges = new RangeSet();
         while (true)
         {
@@ -471,11 +460,11 @@ public class AMQSession_0_10 extends AMQSession<BasicMessageConsumer_0_10, Basic
             {
                 break;
             }
-            ranges.add((int) (long) tag);
+
+            ranges.add(tag.intValue());
         }
-        getQpidSession().messageRelease(ranges, Option.SET_REDELIVERED);
-        // We need to sync so that we get notify of an error.
-        sync();
+
+        return ranges;
     }
 
 
@@ -997,32 +986,26 @@ public class AMQSession_0_10 extends AMQSession<BasicMessageConsumer_0_10, Basic
         }
     }
 
-    @Override
-    public void commit() throws JMSException
+    public void commitImpl() throws AMQException, FailoverException, TransportException
     {
-        checkTransacted();
-        try
-        {
-            if( _txSize > 0 )
-            {
-                messageAcknowledge(_txRangeSet, true);
-                _txRangeSet.clear();
-                _txSize = 0;
-            }
-            sendCommit();
-        }
-        catch(TransportException e)
+        if( _txSize > 0 )
         {
-            throw toJMSException("Session exception occured while trying to commit: " + e.getMessage(), e);
+            messageAcknowledge(_txRangeSet, true);
+            _txRangeSet.clear();
+            _txSize = 0;
         }
-        catch (AMQException e)
+
+        getQpidSession().setAutoSync(true);
+        try
         {
-            throw new JMSAMQException("Failed to commit: " + e.getMessage(), e);
+            getQpidSession().txCommit();
         }
-        catch (FailoverException e)
+        finally
         {
-            throw new JMSAMQException("Fail-over interrupted commit. Status of the commit is uncertain.", e);
+            getQpidSession().setAutoSync(false);
         }
+        // We need to sync so that we get notify of an error.
+        sync();
     }
 
     protected final boolean tagLE(long tag1, long tag2)
@@ -1385,4 +1368,14 @@ public class AMQSession_0_10 extends AMQSession<BasicMessageConsumer_0_10, Basic
         return sb.toString();
     }
 
+    protected void acknowledgeImpl()
+    {
+        RangeSet range = gatherUnackedRangeSet();
+
+        if(range.size() > 0 )
+        {
+            messageAcknowledge(range, true);
+            getQpidSession().sync();
+        }
+    }
 }
diff --git a/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_8.java b/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_8.java
index 3ae6af6350..ccb2b00947 100644
--- a/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_8.java
+++ b/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_8.java
@@ -76,6 +76,7 @@ import org.apache.qpid.framing.amqp_0_91.MethodRegistry_0_91;
 import org.apache.qpid.jms.Session;
 import org.apache.qpid.protocol.AMQConstant;
 import org.apache.qpid.protocol.AMQMethodEvent;
+import org.apache.qpid.transport.TransportException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -125,6 +126,20 @@ public final class AMQSession_0_8 extends AMQSession<BasicMessageConsumer_0_8, B
         return getProtocolHandler().getProtocolVersion();
     }
 
+    protected void acknowledgeImpl()
+    {
+        while (true)
+        {
+            Long tag = _unacknowledgedMessageTags.poll();
+            if (tag == null)
+            {
+                break;
+            }
+
+            acknowledgeMessage(tag, false);
+        }
+    }
+
     public void acknowledgeMessage(long deliveryTag, boolean multiple)
     {
         BasicAckBody body = getMethodRegistry().createBasicAckBody(deliveryTag, multiple);
@@ -170,8 +185,20 @@ public final class AMQSession_0_8 extends AMQSession<BasicMessageConsumer_0_8, B
         }
     }
 
-    public void sendCommit() throws AMQException, FailoverException
+    public void commitImpl() throws AMQException, FailoverException, TransportException
     {
+        // Acknowledge all delivered messages
+        while (true)
+        {
+            Long tag = _deliveredMessageTags.poll();
+            if (tag == null)
+            {
+                break;
+            }
+
+            acknowledgeMessage(tag, false);
+        }
+
         final AMQProtocolHandler handler = getProtocolHandler();
 
         handler.syncWrite(getProtocolHandler().getMethodRegistry().createTxCommitBody().generateFrame(_channelId), TxCommitOkBody.class);
diff --git a/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java b/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
index 0dc8c8c3ed..e6e1398a35 100644
--- a/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
+++ b/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
@@ -37,10 +37,7 @@ import javax.jms.MessageListener;
 import java.util.Arrays;
 import java.util.Iterator;
 import java.util.List;
-import java.util.SortedSet;
 import java.util.ArrayList;
-import java.util.Collections;
-import java.util.TreeSet;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.LinkedBlockingQueue;
@@ -118,29 +115,10 @@ public abstract class BasicMessageConsumer<U> extends Closeable implements Messa
     protected final int _acknowledgeMode;
 
     /**
-     * Number of messages unacknowledged in DUPS_OK_ACKNOWLEDGE mode
-     */
-    private int _outstanding;
-
-    /**
-     * Switch to enable sending of acknowledgements when using DUPS_OK_ACKNOWLEDGE mode. Enabled when _outstannding
-     * number of msgs >= _prefetchHigh and disabled at < _prefetchLow
-     */
-    private boolean _dups_ok_acknowledge_send;
-
-    /**
      * List of tags delievered, The last of which which should be acknowledged on commit in transaction mode.
      */
     private ConcurrentLinkedQueue<Long> _receivedDeliveryTags = new ConcurrentLinkedQueue<Long>();
 
-    /** The last tag that was "multiple" acknowledged on this session (if transacted) */
-    private long _lastAcked;
-
-    /** set of tags which have previously been acked; but not part of the multiple ack (transacted mode only) */
-    private final SortedSet<Long> _previouslyAcked = new TreeSet<Long>();
-
-    private final Object _commitLock = new Object();
-
     /**
      * The thread that was used to call receive(). This is important for being able to interrupt that thread if a
      * receive() is in progress.
@@ -290,17 +268,6 @@ public abstract class BasicMessageConsumer<U> extends Closeable implements Messa
         }
     }
 
-    protected void preApplicationProcessing(AbstractJMSMessage jmsMsg) throws JMSException
-    {
-        if (_session.getAcknowledgeMode() == Session.CLIENT_ACKNOWLEDGE)
-        {
-            _session.addUnacknowledgedMessage(jmsMsg.getDeliveryTag());
-        }
-        
-        _session.setInRecovery(false);
-        preDeliver(jmsMsg);
-    }
-
     /**
      * @param immediate if true then return immediately if the connection is failing over
      *
@@ -409,7 +376,7 @@ public abstract class BasicMessageConsumer<U> extends Closeable implements Messa
             final AbstractJMSMessage m = returnMessageOrThrow(o);
             if (m != null)
             {
-                preApplicationProcessing(m);
+                preDeliver(m);
                 postDeliver(m);
             }
             return m;
@@ -482,7 +449,7 @@ public abstract class BasicMessageConsumer<U> extends Closeable implements Messa
             final AbstractJMSMessage m = returnMessageOrThrow(o);
             if (m != null)
             {
-                preApplicationProcessing(m);
+                preDeliver(m);
                 postDeliver(m);
             }
 
@@ -734,7 +701,7 @@ public abstract class BasicMessageConsumer<U> extends Closeable implements Messa
         {
             if (isMessageListenerSet())
             {
-                preApplicationProcessing(jmsMessage);
+                preDeliver(jmsMessage);
                 getMessageListener().onMessage(jmsMessage);
                 postDeliver(jmsMessage);
             }
@@ -758,19 +725,28 @@ public abstract class BasicMessageConsumer<U> extends Closeable implements Messa
         }
     }
 
-    void preDeliver(AbstractJMSMessage msg)
+    protected void preDeliver(AbstractJMSMessage msg)
     {
+        _session.setInRecovery(false);
+
         switch (_acknowledgeMode)
         {
-
             case Session.PRE_ACKNOWLEDGE:
                 _session.acknowledgeMessage(msg.getDeliveryTag(), false);
                 break;
-
             case Session.CLIENT_ACKNOWLEDGE:
-                // we set the session so that when the user calls acknowledge() it can call the method on session
-                // to send out the appropriate frame
-                msg.setAMQSession(_session);
+                if (isNoConsume())
+                {
+                    _session.acknowledgeMessage(msg.getDeliveryTag(), false);
+                }
+                else
+                {
+                    // we set the session so that when the user calls acknowledge() it can call the method on session
+                    // to send out the appropriate frame
+                    msg.setAMQSession(_session);
+                    _session.addUnacknowledgedMessage(msg.getDeliveryTag());
+                    _session.markDirty();
+                }
                 break;
             case Session.SESSION_TRANSACTED:
                 if (isNoConsume())
@@ -792,15 +768,6 @@ public abstract class BasicMessageConsumer<U> extends Closeable implements Messa
     {
         switch (_acknowledgeMode)
         {
-
-            case Session.CLIENT_ACKNOWLEDGE:
-                if (isNoConsume())
-                {
-                    _session.acknowledgeMessage(msg.getDeliveryTag(), false);
-                }
-                _session.markDirty();
-                break;
-
             case Session.DUPS_OK_ACKNOWLEDGE:
             case Session.AUTO_ACKNOWLEDGE:
                 // we do not auto ack a message if the application code called recover()
diff --git a/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java b/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java
index 47da59724c..d3494298d3 100644
--- a/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java
+++ b/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java
@@ -203,9 +203,11 @@ public class BasicMessageConsumer_0_10 extends BasicMessageConsumer<UnprocessedM
         super.notifyMessage(messageFrame);
     }
 
-    @Override protected void preApplicationProcessing(AbstractJMSMessage jmsMsg) throws JMSException
+    @Override
+    protected void preDeliver(AbstractJMSMessage jmsMsg)
     {
-        super.preApplicationProcessing(jmsMsg);
+        super.preDeliver(jmsMsg);
+
         if (!_session.getTransacted() && _session.getAcknowledgeMode() != org.apache.qpid.jms.Session.CLIENT_ACKNOWLEDGE)
         {
             _session.addUnacknowledgedMessage(jmsMsg.getDeliveryTag());
diff --git a/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java b/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java
index e95f9a7414..bf4de782a5 100644
--- a/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java
+++ b/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java
@@ -455,16 +455,6 @@ public abstract class BasicMessageProducer extends Closeable implements org.apac
 
         AbstractJMSMessage message = convertToNativeMessage(origMessage);
 
-        if (_transacted)
-        {
-            if (_session.hasFailedOver() && _session.isDirty())
-            {
-                throw new JMSAMQException("Failover has occurred and session is dirty so unable to send.",
-                                          new AMQSessionDirtyException("Failover has occurred and session is dirty " +
-                                                                       "so unable to send."));
-            }
-        }
-
         UUID messageId = null;
         if (_disableMessageId)
         {
diff --git a/java/client/src/test/java/org/apache/qpid/test/unit/message/TestAMQSession.java b/java/client/src/test/java/org/apache/qpid/test/unit/message/TestAMQSession.java
index 71a3412dc1..6759b43387 100644
--- a/java/client/src/test/java/org/apache/qpid/test/unit/message/TestAMQSession.java
+++ b/java/client/src/test/java/org/apache/qpid/test/unit/message/TestAMQSession.java
@@ -64,7 +64,12 @@ public class TestAMQSession extends AMQSession<BasicMessageConsumer_0_8, BasicMe
 
     }
 
-    public void sendCommit() throws AMQException, FailoverException
+    public void commitImpl() throws AMQException, FailoverException
+    {
+
+    }
+
+    public void acknowledgeImpl()
     {
 
     }
diff --git a/java/systests/src/main/java/org/apache/qpid/test/client/timeouts/SyncWaitTimeoutDelayTest.java b/java/systests/src/main/java/org/apache/qpid/test/client/timeouts/SyncWaitTimeoutDelayTest.java
index 1a23eee8ab..6189c37306 100644
--- a/java/systests/src/main/java/org/apache/qpid/test/client/timeouts/SyncWaitTimeoutDelayTest.java
+++ b/java/systests/src/main/java/org/apache/qpid/test/client/timeouts/SyncWaitTimeoutDelayTest.java
@@ -63,7 +63,7 @@ public class SyncWaitTimeoutDelayTest extends SyncWaitDelayTest
         catch (JMSException e)
         {
             assertTrue("Wrong exception type received.", e.getLinkedException() instanceof AMQTimeoutException);
-            assertTrue("Wrong message received on exception.", e.getMessage().startsWith("Failed to commit"));
+            assertTrue("Wrong message received on exception.", e.getMessage().startsWith("Exception during commit"));
             // As we are using Nano time ensure to multiply up the millis.            
             assertTrue("Timeout was more than 30s default", (System.nanoTime() - start) < (1000000L * 1000 * 30));
         }
diff --git a/java/systests/src/main/java/org/apache/qpid/test/unit/ack/ClientAcknowledgeTest.java b/java/systests/src/main/java/org/apache/qpid/test/unit/ack/ClientAcknowledgeTest.java
new file mode 100644
index 0000000000..06be5cf456
--- /dev/null
+++ b/java/systests/src/main/java/org/apache/qpid/test/unit/ack/ClientAcknowledgeTest.java
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.qpid.test.unit.ack;
+
+import javax.jms.Connection;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Queue;
+import javax.jms.Session;
+
+import org.apache.qpid.test.utils.QpidBrokerTestCase;
+
+public class ClientAcknowledgeTest extends QpidBrokerTestCase
+{
+    private static final long ONE_DAY_MS = 1000l * 60 * 60 * 24;
+    private Connection _connection;
+    private Queue _queue;
+    private Session _consumerSession;
+    private MessageConsumer _consumer;
+    private MessageProducer _producer;
+
+    @Override
+    protected void setUp() throws Exception
+    {
+        super.setUp();
+        _queue = getTestQueue();
+        _connection = getConnection();
+    }
+
+    /**
+     * Test that message.acknowledge actually acknowledges, regardless of
+     * the flusher thread period, by restarting the broker after calling
+     * acknowledge, and then verifying after restart that the message acked
+     * is no longer present. This test requires a persistent store.
+     */
+    public void testClientAckWithLargeFlusherPeriod() throws Exception
+    {
+        setTestClientSystemProperty("qpid.session.max_ack_delay", Long.toString(ONE_DAY_MS));
+        _consumerSession = _connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
+        _consumer = _consumerSession.createConsumer(_queue);
+        _connection.start();
+
+        _producer = _consumerSession.createProducer(_queue);
+        _producer.send(createNextMessage(_consumerSession, 1));
+        _producer.send(createNextMessage(_consumerSession, 2));
+
+        Message message = _consumer.receive(1000l);
+        assertNotNull("Message has not been received", message);
+        assertEquals("Unexpected message is received", 1, message.getIntProperty(INDEX));
+        message.acknowledge();
+
+        //restart broker to allow verification of the acks
+        //without explicitly closing connection (which acks)
+        restartBroker();
+
+        // try to receive the message again, which should fail (as it was ackd)
+        _connection = getConnection();
+        _connection.start();
+        _consumerSession = _connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
+        _consumer = _consumerSession.createConsumer(_queue);
+        message = _consumer.receive(1000l);
+        assertNotNull("Message has not been received", message);
+        assertEquals("Unexpected message is received", 2, message.getIntProperty(INDEX));
+    }
+}
diff --git a/java/systests/src/main/java/org/apache/qpid/test/unit/publish/DirtyTransactedPublishTest.java b/java/systests/src/main/java/org/apache/qpid/test/unit/publish/DirtyTransactedPublishTest.java
deleted file mode 100644
index 3ec7937812..0000000000
--- a/java/systests/src/main/java/org/apache/qpid/test/unit/publish/DirtyTransactedPublishTest.java
+++ /dev/null
@@ -1,403 +0,0 @@
-/*
- *
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *   http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- *
- */
-package org.apache.qpid.test.unit.publish;
-
-import org.apache.qpid.client.AMQConnection;
-import org.apache.qpid.client.AMQDestination;
-import org.apache.qpid.client.AMQSession;
-import org.apache.qpid.jms.ConnectionListener;
-import org.apache.qpid.test.utils.FailoverBaseCase;
-
-import javax.jms.Connection;
-import javax.jms.JMSException;
-import javax.jms.Message;
-import javax.jms.MessageConsumer;
-import javax.jms.MessageListener;
-import javax.jms.MessageProducer;
-import javax.jms.Queue;
-import javax.jms.Session;
-import javax.jms.TransactionRolledBackException;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicReference;
-
-/**
- * QPID-1816 : Whilst testing Acknoledgement after failover this completes testing
- * of the client after failover. When we have a dirty session we should receive
- * an error if we attempt to publish. This test ensures that both in the synchronous
- * and asynchronous message delivery paths we receive the expected exceptions at
- * the expected time.
- */
-public class DirtyTransactedPublishTest extends FailoverBaseCase implements ConnectionListener
-{
-    protected CountDownLatch _failoverCompleted = new CountDownLatch(1);
-
-    protected int NUM_MESSAGES;
-    protected Connection _connection;
-    protected Queue _queue;
-    protected Session _consumerSession;
-    protected MessageConsumer _consumer;
-    protected MessageProducer _producer;
-
-    private static final String MSG = "MSG";
-    private static final String SEND_FROM_ON_MESSAGE_TEXT = "sendFromOnMessage";
-    protected CountDownLatch _receviedAll;
-    protected AtomicReference<Exception> _causeOfFailure = new AtomicReference<Exception>(null);
-
-    @Override
-    protected void setUp() throws Exception
-    {
-        super.setUp();
-        NUM_MESSAGES = 10;
-
-        _queue = getTestQueue();
-
-        //Create Producer put some messages on the queue
-        _connection = getConnection();
-    }
-
-    /**
-     * Initialise the test variables
-     * @param transacted is this a transacted test
-     * @param mode if not trasacted then what ack mode to use
-     * @throws Exception if there is a setup issue.
-     */
-    protected void init(boolean transacted, int mode) throws Exception
-    {
-        _consumerSession = _connection.createSession(transacted, mode);
-        _consumer = _consumerSession.createConsumer(_queue);
-        _producer = _consumerSession.createProducer(_queue);
-
-        // These should all end up being prefetched by session
-        sendMessage(_consumerSession, _queue, 1);
-
-        assertEquals("Wrong number of messages on queue", 1,
-                     ((AMQSession) _consumerSession).getQueueDepth((AMQDestination) _queue));
-    }
-
-    /**
-     * If a transacted session has failed over whilst it has uncommitted sent
-     * data then we need to throw a TransactedRolledbackException on commit()
-     *
-     * The alternative would be to maintain a replay buffer so that the message
-     * could be resent. This is not currently implemented
-     *
-     * @throws Exception if something goes wrong.
-     */
-    public void testDirtySendingSynchronousTransacted() throws Exception
-    {
-        Session producerSession = _connection.createSession(true, Session.SESSION_TRANSACTED);
-
-        // Ensure we get failover notifications
-        ((AMQConnection) _connection).setConnectionListener(this);
-
-        MessageProducer producer = producerSession.createProducer(_queue);
-
-        // Create and send message 0
-        Message msg = producerSession.createMessage();
-        msg.setIntProperty(INDEX, 0);
-        producer.send(msg);
-
-        // DON'T commit message .. fail connection
-
-        failBroker(getFailingPort());
-
-        // Ensure destination exists for sending
-        producerSession.createConsumer(_queue).close();
-
-        // Send the next message
-        msg.setIntProperty(INDEX, 1);
-        try
-        {
-            producer.send(msg);
-            fail("Should fail with Qpid as we provide early warning of the dirty session via a JMSException.");
-        }
-        catch (JMSException jmse)
-        {
-            assertEquals("Early warning of dirty session not correct",
-                         "Failover has occurred and session is dirty so unable to send.", jmse.getMessage());
-        }
-
-        // Ignore that the session is dirty and attempt to commit to validate the
-        // exception is thrown. AND that the above failure notification did NOT
-        // clean up the session.
-
-        try
-        {
-            producerSession.commit();
-            fail("Session is dirty we should get an TransactionRolledBackException");
-        }
-        catch (TransactionRolledBackException trbe)
-        {
-            // Normal path.
-        }
-
-        // Resending of messages should now work ok as the commit was forcilbly rolledback
-        msg.setIntProperty(INDEX, 0);
-        producer.send(msg);
-        msg.setIntProperty(INDEX, 1);
-        producer.send(msg);
-
-        producerSession.commit();
-
-        assertEquals("Wrong number of messages on queue", 2,
-                     ((AMQSession) producerSession).getQueueDepth((AMQDestination) _queue));
-    }
-
-    /**
-     * If a transacted session has failed over whilst it has uncommitted sent
-     * data then we need to throw a TransactedRolledbackException on commit()
-     *
-     * The alternative would be to maintain a replay buffer so that the message
-     * could be resent. This is not currently implemented
-     *
-     * @throws Exception if something goes wrong.
-     */
-    public void testDirtySendingOnMessageTransacted() throws Exception
-    {
-        NUM_MESSAGES = 1;
-        _receviedAll = new CountDownLatch(NUM_MESSAGES);
-        ((AMQConnection) _connection).setConnectionListener(this);
-
-        init(true, Session.SESSION_TRANSACTED);
-
-        _consumer.setMessageListener(new MessageListener()
-        {
-
-            public void onMessage(Message message)
-            {
-                try
-                {
-                    // Create and send message 0
-                    Message msg = _consumerSession.createMessage();
-                    msg.setIntProperty(INDEX, 0);
-                    _producer.send(msg);
-
-                    // DON'T commit message .. fail connection
-
-                    failBroker(getFailingPort());
-
-                    // rep
-                    repopulateBroker();
-
-                    // Destination will exist as this failBroker will populate
-                    // the queue with 1 message
-
-                    // Send the next message
-                    msg.setIntProperty(INDEX, 1);
-                    try
-                    {
-                        _producer.send(msg);
-                        fail("Should fail with Qpid as we provide early warning of the dirty session via a JMSException.");
-                    }
-                    catch (JMSException jmse)
-                    {
-                        assertEquals("Early warning of dirty session not correct",
-                                     "Failover has occurred and session is dirty so unable to send.", jmse.getMessage());
-                    }
-
-                    // Ignore that the session is dirty and attempt to commit to validate the
-                    // exception is thrown. AND that the above failure notification did NOT
-                    // clean up the session.
-
-                    try
-                    {
-                        _consumerSession.commit();
-                        fail("Session is dirty we should get an TransactionRolledBackException");
-                    }
-                    catch (TransactionRolledBackException trbe)
-                    {
-                        // Normal path.
-                    }
-
-                    // Resend messages
-                    msg.setIntProperty(INDEX, 0);
-                    msg.setStringProperty(MSG, SEND_FROM_ON_MESSAGE_TEXT);
-                    _producer.send(msg);
-                    msg.setIntProperty(INDEX, 1);
-                    msg.setStringProperty(MSG, SEND_FROM_ON_MESSAGE_TEXT);
-                    _producer.send(msg);
-
-                    _consumerSession.commit();
-
-                    // Stop this consumer .. can't do _consumer.stop == DEADLOCK
-                    // this doesn't seem to stop dispatcher running
-                    _connection.stop();
-
-                    // Signal that the onMessage send part of test is complete
-                    // main thread can validate that messages are correct
-                    _receviedAll.countDown();
-
-                }
-                catch (Exception e)
-                {
-                    fail(e);
-                }
-
-            }
-
-        });
-
-        _connection.start();
-
-        if (!_receviedAll.await(10000L, TimeUnit.MILLISECONDS))
-        {
-            // Check to see if we ended due to an exception in the onMessage handler
-            Exception cause = _causeOfFailure.get();
-            if (cause != null)
-            {
-                cause.printStackTrace();
-                fail(cause.getMessage());
-            }
-            else
-            {
-                fail("All messages not received:" + _receviedAll.getCount() + "/" + NUM_MESSAGES);
-            }
-        }
-
-        // Check to see if we ended due to an exception in the onMessage handler
-        Exception cause = _causeOfFailure.get();
-        if (cause != null)
-        {
-            cause.printStackTrace();
-            fail(cause.getMessage());
-        }
-
-        _consumer.close();
-        _consumerSession.close();
-
-        _consumerSession = _connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
-
-        _connection.start();
-
-        // Validate that we could send the messages as expected.
-        assertEquals("Wrong number of messages on queue", 3,
-                     ((AMQSession) _consumerSession).getQueueDepth((AMQDestination) _queue));
-
-        MessageConsumer consumer = _consumerSession.createConsumer(_queue);
-
-        //Validate the message sent to setup the failed over broker.
-        Message message = consumer.receive(1000);
-        assertNotNull("Message " + 0 + " not received.", message);
-        assertEquals("Incorrect message received", 0, message.getIntProperty(INDEX));
-
-        // Validate the two messages sent from within the onMessage
-        for (int index = 0; index <= 1; index++)
-        {
-            message = consumer.receive(1000);
-            assertNotNull("Message " + index + " not received.", message);
-            assertEquals("Incorrect message received", index, message.getIntProperty(INDEX));
-            assertEquals("Incorrect message text for message:" + index, SEND_FROM_ON_MESSAGE_TEXT, message.getStringProperty(MSG));
-        }
-
-        assertNull("Extra message received.", consumer.receiveNoWait());
-
-        _consumerSession.close();
-
-        assertEquals("Wrong number of messages on queue", 0,
-                     ((AMQSession) getConnection().createSession(false, Session.AUTO_ACKNOWLEDGE)).getQueueDepth((AMQDestination) _queue));
-    }
-
-    private void repopulateBroker() throws Exception
-    {
-        // Repopulate this new broker so we can test what happends after failover
-
-        //Get the connection to the first (main port) broker.
-        Connection connection = getConnection();
-        // Use a transaction to send messages so we can be sure they arrive.
-        Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
-        // ensure destination is created.
-        session.createConsumer(_queue).close();
-
-        sendMessage(session, _queue, NUM_MESSAGES);
-
-        assertEquals("Wrong number of messages on queue", NUM_MESSAGES,
-                     ((AMQSession) session).getQueueDepth((AMQDestination) _queue));
-
-        connection.close();
-    }
-
-    // AMQConnectionListener Interface.. used so we can validate that we
-    // actually failed over.
-
-    public void bytesSent(long count)
-    {
-    }
-
-    public void bytesReceived(long count)
-    {
-    }
-
-    public boolean preFailover(boolean redirect)
-    {
-        //Allow failover
-        return true;
-    }
-
-    public boolean preResubscribe()
-    {
-        //Allow failover
-        return true;
-    }
-
-    public void failoverComplete()
-    {
-        _failoverCompleted.countDown();
-    }
-
-    /**
-     * Override so we can block until failover has completd
-     *
-     * @param port int the port of the broker to fail.
-     */
-    @Override
-    public void failBroker(int port)
-    {
-        super.failBroker(port);
-
-        try
-        {
-            if (!_failoverCompleted.await(DEFAULT_FAILOVER_TIME, TimeUnit.MILLISECONDS))
-            {
-                fail("Failover did not occur in specified time:" + DEFAULT_FAILOVER_TIME);
-            }
-        }
-        catch (InterruptedException e)
-        {
-            fail("Failover was interuppted");
-        }
-    }
-
-    /**
-     * Pass the given exception back to the waiting thread to fail the test run.
-     *
-     * @param e The exception that is causing the test to fail.
-     */
-    protected void fail(Exception e)
-    {
-        _causeOfFailure.set(e);
-        // End the test.
-        while (_receviedAll.getCount() != 0)
-        {
-            _receviedAll.countDown();
-        }
-    }
-}
diff --git a/java/test-profiles/CPPExcludes b/java/test-profiles/CPPExcludes
index 2b58a0684d..d05f42e4b7 100755
--- a/java/test-profiles/CPPExcludes
+++ b/java/test-profiles/CPPExcludes
@@ -142,7 +142,6 @@ org.apache.qpid.server.failover.MessageDisappearWithIOExceptionTest#*
 
 // These are recent test additions that are failing with the c++ broker
 // Temporarily disabling until properly investigated.
-org.apache.qpid.test.unit.publish.DirtyTransactedPublishTest#*
 org.apache.qpid.test.unit.ack.FailoverBeforeConsumingRecoverTest#*
 
 org.apache.qpid.test.client.RollbackOrderTest#testOrderingAfterRollbackOnMessage#*
diff --git a/java/test-profiles/CPPTransientExcludes b/java/test-profiles/CPPTransientExcludes
index 47f24db19c..a214cf5b5c 100644
--- a/java/test-profiles/CPPTransientExcludes
+++ b/java/test-profiles/CPPTransientExcludes
@@ -27,3 +27,6 @@ org.apache.qpid.test.unit.xa.TopicTest#testMultiMessagesDurSubCrash
 org.apache.qpid.test.unit.xa.TopicTest#testRecover
 org.apache.qpid.test.unit.xa.QueueTest#testRecover
 org.apache.qpid.test.unit.xa.QueueTest#testSendAndRecover
+
+// test requires a persistent store
+org.apache.qpid.test.unit.ack.ClientAcknowledgeTest#testClientAckWithLargeFlusherPeriod
diff --git a/java/test-profiles/Java010Excludes b/java/test-profiles/Java010Excludes
index fe0a53bdfc..e7718b982d 100755
--- a/java/test-profiles/Java010Excludes
+++ b/java/test-profiles/Java010Excludes
@@ -55,9 +55,6 @@ org.apache.qpid.server.queue.ProducerFlowControlTest#*
 //QPID-1950 : Commit to test this failure. This is a MINA only failure so it cannot be tested when using 010.
 org.apache.qpid.server.failover.MessageDisappearWithIOExceptionTest#*
 
-//QPID-3421: tests are failing on 0.10 test profile
-org.apache.qpid.test.unit.publish.DirtyTransactedPublishTest#*
-
 //QPID-1864: rollback with subscriptions does not work in 0-10 yet
 org.apache.qpid.test.client.RollbackOrderTest#testOrderingAfterRollbackOnMessage
 
diff --git a/java/test-profiles/JavaTransientExcludes b/java/test-profiles/JavaTransientExcludes
index 2ce514461a..67190a6fcc 100644
--- a/java/test-profiles/JavaTransientExcludes
+++ b/java/test-profiles/JavaTransientExcludes
@@ -19,6 +19,7 @@
 
 //These tests require a persistent store
 org.apache.qpid.server.store.PersistentStoreTest#*
+org.apache.qpid.test.unit.ack.ClientAcknowledgeTest#testClientAckWithLargeFlusherPeriod
 
 org.apache.qpid.test.unit.ct.DurableSubscriberTest#*