Merged revisions 598285,598619,598721,598834-598835,599375,599531,599533,599572,599805,602134,604151,604928,605536,605542,606015-606016 via svnmerge from

https://svn.apache.org/repos/asf/incubator/qpid/branches/M2.1.1

........
  r598285 | ritchiem | 2007-11-26 14:16:01 +0000 (Mon, 26 Nov 2007) | 3 lines
  
  QPID-92, QPID-564 : Upgraded Mina to 1.0.1 still not good enough but all future versions currently have a bug with the CumulativeProtocolDecoder. It compact()s the buffer which breaks slices. Added MultiThread Support which is some of the feature set of QPID-564
........
  r598619 | ritchiem | 2007-11-27 12:51:14 +0000 (Tue, 27 Nov 2007) | 1 line
  
  Renamed POMs to M2.1.1 Removed erroneous equals() in SpecificMethodFrameListenerTest
........
  r598721 | ritchiem | 2007-11-27 18:09:33 +0000 (Tue, 27 Nov 2007) | 1 line
  
  QPID-621 : Patch Supplied by Aidan Skinner. Msg Ack after msg consumer is closed.
........
  r598834 | ritchiem | 2007-11-28 00:45:32 +0000 (Wed, 28 Nov 2007) | 14 lines
  
  QPID-679 : Patch provided by Aidan Skinner and additional from odd problems during test runs.
  AMQChannel - Catch and log AMQException occuring when requeue()-ing. Previously exceptions wouldn't be caught at all. The requeue() is called during closure so there is nothing we can do protocol wise on error other than log the issue and continue with any other shutdown that is needed.
  AMQMinaProtocolSession & AMQPFastProtocolHandler . Additions to catch and log AMQExceptions. Changes to AMQMinaProtocolSession were done to ignore all input on a closing session other than the close-ok. Previously only Protocol frames were ignored this resulted in Content*Body-s still being processed. Additional checks were made for the MessageStoreClosedException to log and continue. As said else were we need to seperate protocol exceptoions(AMQException) from internal code exception handling. Further All AMQExceptions occuring in the frameReceived method are now caught and logged. Allowing them to propogate higher will only result in thread death.
  AMQPFastProtocolHandler Caught AMQExceptions occuring whilst closing the session. Again allowing these to continue will result in thread death. There is not a lot that can be done other than log the problem as the session is already closed by this point. Prevented the stacktrace associated with a session exception being printed in the exceptionCaught method when the problem was an IO Exception. This doesn't add anything useful and only adds to the log file sizes.
  ApplicationRegistry - Added removeAll option which ensures that all ARs are correctly purged so that we can attempt to clean up between Unit Tests.
  MemoryMessageStore - This was causing us real problems during the failover testing. Similar checks should probably be made to any other Message Store Impl. The issue was that when shutting down the broker the MS.close() method is called this sets all the storage to null. However, there may still be message processing going on as the close() does not attempt to stop connection processing. Hence we now check to see if the Store is close throwing a MSClosedException if required. This prevents NPEs that have been seen during Unit failover testing. In fact the close() is called as a request to shutdown the ApplicationRegistry, but this only occurs from tests and broker shutdown, no attempt to unbind or prevent further connections during this period is yet done.
  
  CLIENT CHANGES
  AMQConnection - Added method to check if failover is in progress.
  AMQClient - Upgraded acknowledge() exception to JMSException for errors due to failover. Also , added call to update consumers as a result of failover.
  BasicMessageConsumer - Changes to acquireReceiving to take in to consideration blocking for failover to occur. wrt receiveNoWait.. which previously blocked for failover to complete... not exactly noWait. acknowledge will now
  TransportConnection - Update to ensure all inVM brokers are correctly killed.
  
  FailoverTest - QPID-679 - Finder of all the above problems.
........
  r598835 | ritchiem | 2007-11-28 01:01:05 +0000 (Wed, 28 Nov 2007) | 1 line
  
  CommitRollbackTest - this one just was never right.. now we have something better.
........
  r599375 | ritchiem | 2007-11-29 10:58:08 +0000 (Thu, 29 Nov 2007) | 1 line
  
  Update to broker to address fanout python failure.
........
  r599531 | ritchiem | 2007-11-29 17:56:12 +0000 (Thu, 29 Nov 2007) | 1 line
  
  QPID-92 QPID-564 Forgot to upgrade mina to 1.0.1
........
  r599533 | ritchiem | 2007-11-29 18:25:21 +0000 (Thu, 29 Nov 2007) | 1 line
  
  QPID-564 QPID-92 Tidied up a few points and fixed infinite loop in Read IO Thread
........
  r599572 | ritchiem | 2007-11-29 20:56:22 +0000 (Thu, 29 Nov 2007) | 2 lines
  
  Mina Fix: Vm Pipe Starts Connection session before acceptor session. This results in protocol frames arriving before the protocol decoder has been configured on the InVM Broker. Verification of this could be done by adding a client side filter that delays the first message by a few seconds.
........
  r599805 | ritchiem | 2007-11-30 12:47:08 +0000 (Fri, 30 Nov 2007) | 1 line
  
  Added new simple Request/Repsonse code as my last commit here seems to have missed the actual code.
........
  r602134 | rupertlssmith | 2007-12-07 16:00:14 +0000 (Fri, 07 Dec 2007) | 1 line
  
  Added JDNI config for two broker, failover setup for failover tests. Also passed into FT tests config.
........
  r604151 | ritchiem | 2007-12-14 10:40:37 +0000 (Fri, 14 Dec 2007) | 2 lines
  
  QPID-707 : Added new test to check message count on broker as messages are consumed to ensure that an ack is sent at 5000 mgs. Added acks on message consumer closure.
  Augmented VMTestCase to have helper methods for accessing broker statistics.
........
  r604928 | rupertlssmith | 2007-12-17 17:00:10 +0000 (Mon, 17 Dec 2007) | 1 line
  
  DUPS_OK mode set to be same as AUTO_ACK, fixed broken dups ok test.
........
  r605536 | rupertlssmith | 2007-12-19 13:40:05 +0000 (Wed, 19 Dec 2007) | 1 line
  
  Messages were being sent mandatory by default, set to false.
........
  r605542 | rupertlssmith | 2007-12-19 13:53:44 +0000 (Wed, 19 Dec 2007) | 1 line
  
  Changed test configs to use colons instead of commas.
........
  r606015 | rgodfrey | 2007-12-20 20:08:01 +0000 (Thu, 20 Dec 2007) | 2 lines
  
  QPID-714 : (Patch from Aidan Skinner) Issue with competing, transactional/client-ack consumers
  Ack each individual message on commit, not use multiple acks
........
  r606016 | rgodfrey | 2007-12-20 20:12:25 +0000 (Thu, 20 Dec 2007) | 2 lines
  
  QPID-714 : (Patch from Aidan Skinner) Issue with competing, transactional/client-ack consumers
  Ack each individual message on commit, not use multiple acks
........


git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/branches/M2.1@614906 13f79535-47bb-0310-9956-ffa450edef68

10 files changed, 152 insertions, 197 deletions

diff --git a/java/client/example/src/main/java/org/apache/qpid/example/simple/reqresp/Client.java b/java/client/example/src/main/java/org/apache/qpid/example/simple/reqresp/Client.java
index 02386e84eb..b6badff24d 100644
--- a/java/client/example/src/main/java/org/apache/qpid/example/simple/reqresp/Client.java
+++ b/java/client/example/src/main/java/org/apache/qpid/example/simple/reqresp/Client.java
@@ -3,7 +3,7 @@
  * 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
+  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
@@ -252,4 +252,3 @@ public class Client implements MessageListener
         new Client();
     }
 }
-
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 38325a1e41..39b3b80e74 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
@@ -1303,4 +1303,8 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
         _protocolHandler.getProtocolSession().setProtocolVersion(protocolVersion);
     }
 
+    public boolean isFailingOver()
+    {
+        return (_protocolHandler.getFailoverLatch() != null);
+    }
 }
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 15c113a05d..42f07f97f9 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
@@ -83,6 +83,7 @@ import java.util.Arrays;
 import java.util.Iterator;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -219,6 +220,12 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi
     private Map<AMQShortString, BasicMessageConsumer> _consumers =
             new ConcurrentHashMap<AMQShortString, BasicMessageConsumer>();
 
+    /**
+     * Contains a list of consumers which have been removed but which might still have
+     * messages to acknowledge, eg in client ack or transacted modes
+     */
+    private CopyOnWriteArrayList<BasicMessageConsumer> _removedConsumers = new CopyOnWriteArrayList<BasicMessageConsumer>();
+
     /** Provides a count of consumers on destinations, in order to be able to know if a destination has consumers. */
     private ConcurrentHashMap<Destination, AtomicInteger> _destinationConsumerCount =
             new ConcurrentHashMap<Destination, AtomicInteger>();
@@ -387,7 +394,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi
      *
      * @throws IllegalStateException If the session is closed.
      */
-    public void acknowledge() throws IllegalStateException
+    public void acknowledge() throws JMSException
     {
         if (isClosed())
         {
@@ -611,20 +618,19 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi
 
                     for (Iterator<BasicMessageConsumer> i = _consumers.values().iterator(); i.hasNext();)
                     {
-//                        i.next().acknowledgeLastDelivered();
-//                    }
-
-                        // get next acknowledgement to server
-                        Long next = i.next().getLastDelivered();
-                        if (next != null && next > lastTag)
-                        {
-                            lastTag = next;
-                        }
+                        i.next().acknowledgeDelivered();
                     }
 
-                    if (lastTag != -1)
+                    if (_transacted)
                     {
-                        acknowledgeMessage(lastTag, true);
+                        // Do the above, but for consumers which have been de-registered since the
+                        // last commit
+                        for (int i = 0; i < _removedConsumers.size(); i++)
+                        {
+                            // Sends acknowledgement to server
+                            _removedConsumers.get(i).acknowledgeDelivered();
+                            _removedConsumers.remove(i);
+                        }
                     }
 
                     // Commits outstanding messages sent and outstanding acknowledgements.
@@ -760,7 +766,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi
     {
         checkValidDestination(destination);
 
-        return createConsumerImpl(destination, prefetch, prefetch, noLocal, exclusive, selector, null, false, false);
+        return createConsumerImpl(destination, prefetch, prefetch / 2, noLocal, exclusive, selector, null, false, false);
     }
 
     public MessageConsumer createConsumer(Destination destination, int prefetchHigh, int prefetchLow, boolean noLocal,
@@ -776,7 +782,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi
     {
         checkValidDestination(destination);
 
-        return createConsumerImpl(destination, prefetch, prefetch, noLocal, exclusive, selector, rawSelector, false, false);
+        return createConsumerImpl(destination, prefetch, prefetch / 2, noLocal, exclusive, selector, rawSelector, false, false);
     }
 
     public MessageConsumer createConsumer(Destination destination, int prefetchHigh, int prefetchLow, boolean noLocal,
@@ -1676,6 +1682,13 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi
                     _destinationConsumerCount.remove(dest);
                 }
             }
+
+            // Consumers that are closed in a transaction must be stored
+            // so that messages they have received can be acknowledged on commit
+            if (_transacted)
+            {
+                _removedConsumers.add(consumer);
+            }
         }
     }
 
@@ -2445,6 +2458,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi
         for (Iterator it = consumers.iterator(); it.hasNext();)
         {
             BasicMessageConsumer consumer = (BasicMessageConsumer) it.next();
+            consumer.failedOver();
             registerConsumer(consumer, true);
         }
     }
@@ -2543,17 +2557,13 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi
         return _messageDeliveryLock;
     }
 
-    /**
-     * Signifies that the session has pending sends to commit.
-     */
+    /** Signifies that the session has pending sends to commit. */
     public void markDirty()
     {
         _dirty = true;
     }
 
-    /**
-     * Signifies that the session has no pending sends to commit.
-     */
+    /** Signifies that the session has no pending sends to commit. */
     public void markClean()
     {
         _dirty = false;
@@ -2562,6 +2572,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi
 
     /**
      * Check to see if failover has occured since the last call to markClean(commit or rollback).
+     *
      * @return boolean true if failover has occured.
      */
     public boolean hasFailedOver()
@@ -2571,6 +2582,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi
 
     /**
      * 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()
@@ -2624,7 +2636,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi
                 }
 
                 // Reject messages on pre-receive queue
-                consumer.rollback();
+                consumer.rollbackPendingMessages();
 
                 // Reject messages on pre-dispatch queue
                 rejectMessagesForConsumerTag(consumer.getConsumerTag(), true);
@@ -2668,6 +2680,13 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi
 
                 }
 
+                for (int i = 0; i < _removedConsumers.size(); i++)
+                {
+                    // Sends acknowledgement to server
+                    _removedConsumers.get(i).rollback();
+                    _removedConsumers.remove(i);
+                }
+
                 setConnectionStopped(isStopped);
             }
 
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 ae31f5ebdd..610e0109b1 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
@@ -33,14 +33,12 @@ import org.apache.qpid.framing.BasicCancelOkBody;
 import org.apache.qpid.framing.FieldTable;
 import org.apache.qpid.jms.MessageConsumer;
 import org.apache.qpid.jms.Session;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import javax.jms.JMSException;
 import javax.jms.Message;
 import javax.jms.MessageListener;
-
 import java.util.Arrays;
 import java.util.Iterator;
 import java.util.List;
@@ -255,6 +253,10 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
 
         switch (_acknowledgeMode)
         {
+            case Session.DUPS_OK_ACKNOWLEDGE:
+                _logger.info("Recording tag for acking on close:" + msg.getDeliveryTag());
+                _receivedDeliveryTags.add(msg.getDeliveryTag());
+                break;
 
             case Session.CLIENT_ACKNOWLEDGE:
                 _unacknowledgedDeliveryTags.add(msg.getDeliveryTag());
@@ -277,8 +279,28 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
         _session.setInRecovery(false);
     }
 
-    private void acquireReceiving() throws JMSException
+    /**
+     * @param immediate if true then return immediately if the connection is failing over
+     *
+     * @return boolean if the acquisition was successful
+     *
+     * @throws JMSException
+     * @throws InterruptedException
+     */
+    private boolean acquireReceiving(boolean immediate) throws JMSException, InterruptedException
     {
+        if (_connection.isFailingOver())
+        {
+            if (immediate)
+            {
+                return false;
+            }
+            else
+            {
+                _connection.blockUntilNotFailingOver();
+            }
+        }
+
         if (!_receiving.compareAndSet(false, true))
         {
             throw new javax.jms.IllegalStateException("Another thread is already receiving.");
@@ -290,6 +312,7 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
         }
 
         _receivingThread = Thread.currentThread();
+        return true;
     }
 
     private void releaseReceiving()
@@ -343,7 +366,18 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
 
         checkPreConditions();
 
-        acquireReceiving();
+        try
+        {
+            acquireReceiving(false);
+        }
+        catch (InterruptedException e)
+        {
+            _logger.warn("Interrupted: " + e);
+            if (isClosed())
+            {
+                return null;
+            }
+        }
 
         _session.startDistpatcherIfNecessary();
 
@@ -424,7 +458,25 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
     {
         checkPreConditions();
 
-        acquireReceiving();
+        try
+        {
+            if (!acquireReceiving(true))
+            {
+                //If we couldn't acquire the receiving thread then return null.
+                // This will occur if failing over.
+                return null;
+            }
+        }
+        catch (InterruptedException e)
+        {
+            /*
+             *  This seems slightly shoddy but should never actually be executed
+             *  since we told acquireReceiving to return immediately and it shouldn't
+             *  block on anything.
+             */
+
+            return null;
+        }
 
         _session.startDistpatcherIfNecessary();
 
@@ -721,12 +773,13 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
                 break;
 
             case Session.DUPS_OK_ACKNOWLEDGE:
-                if (++_outstanding >= _prefetchHigh)
+            /*(    if (++_outstanding >= _prefetchHigh)
                 {
                     _dups_ok_acknowledge_send = true;
                 }
 
-                if (_outstanding <= _prefetchLow)
+                //Can't use <= as _prefetchHigh may equal _prefetchLow so no acking would occur.
+                if (_outstanding < _prefetchLow)
                 {
                     _dups_ok_acknowledge_send = false;
                 }
@@ -736,11 +789,12 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
                     if (!_session.isInRecovery())
                     {
                         _session.acknowledgeMessage(msg.getDeliveryTag(), true);
+                        _outstanding = 0;
                     }
                 }
 
                 break;
-
+             */
             case Session.AUTO_ACKNOWLEDGE:
                 // we do not auto ack a message if the application code called recover()
                 if (!_session.isInRecovery())
@@ -777,20 +831,11 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
     }
 
     /** Acknowledge up to last message delivered (if any). Used when commiting. */
-    void acknowledgeLastDelivered()
+    void acknowledgeDelivered()
     {
-        if (!_receivedDeliveryTags.isEmpty())
+    	while (!_receivedDeliveryTags.isEmpty())
         {
-            long lastDeliveryTag = _receivedDeliveryTags.poll();
-
-            while (!_receivedDeliveryTags.isEmpty())
-            {
-                lastDeliveryTag = _receivedDeliveryTags.poll();
-            }
-
-            assert _receivedDeliveryTags.isEmpty();
-
-            _session.acknowledgeMessage(lastDeliveryTag, true);
+    		_session.acknowledgeMessage(_receivedDeliveryTags.poll(), false);
         }
     }
 
@@ -866,11 +911,18 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
         }
     }
 
-    public void acknowledge() // throws JMSException
+    public void acknowledge() throws JMSException
     {
-        if (!isClosed())
+        if (isClosed())
+        {
+            throw new IllegalStateException("Consumer is closed");
+        }
+        else if (_session.hasFailedOver())
+        {
+            throw new JMSException("has failed over");
+        }
+        else
         {
-
             Iterator<Long> tags = _unacknowledgedDeliveryTags.iterator();
             while (tags.hasNext())
             {
@@ -878,10 +930,6 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
                 tags.remove();
             }
         }
-        else
-        {
-            throw new IllegalStateException("Consumer is closed");
-        }
     }
 
     /** Called on recovery to reset the list of delivery tags */
@@ -951,7 +999,11 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
             }
         }
 
-        // rollback pending messages
+        rollbackPendingMessages();
+    }
+
+    public void rollbackPendingMessages()
+    {
         if (_synchronousQueue.size() > 0)
         {
             if (_logger.isDebugEnabled())
@@ -1016,4 +1068,11 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
     {
         _synchronousQueue.clear();
     }
+
+    /** to be called when a failover has occured */
+    public void failedOver()
+    {
+        clearReceiveQueue();
+        clearUnackedMessages();
+    }
 }
diff --git a/java/client/src/main/java/org/apache/qpid/client/protocol/BlockingMethodFrameListener.java b/java/client/src/main/java/org/apache/qpid/client/protocol/BlockingMethodFrameListener.java
index 1badbb601c..2b63475d71 100644
--- a/java/client/src/main/java/org/apache/qpid/client/protocol/BlockingMethodFrameListener.java
+++ b/java/client/src/main/java/org/apache/qpid/client/protocol/BlockingMethodFrameListener.java
@@ -294,18 +294,4 @@ public abstract class BlockingMethodFrameListener implements AMQMethodListener
         }
     }
 
-    public boolean equals(Object o)
-    {
-
-        if (o instanceof BlockingMethodFrameListener)
-        {
-            BlockingMethodFrameListener other = (BlockingMethodFrameListener) o;
-
-            return _channelId == other._channelId;
-        }
-
-        return false;
-    }
-
-
 }
diff --git a/java/client/src/main/java/org/apache/qpid/client/state/listener/SpecificMethodFrameListener.java b/java/client/src/main/java/org/apache/qpid/client/state/listener/SpecificMethodFrameListener.java
index 4a4f4a0a38..c66603b7a0 100644
--- a/java/client/src/main/java/org/apache/qpid/client/state/listener/SpecificMethodFrameListener.java
+++ b/java/client/src/main/java/org/apache/qpid/client/state/listener/SpecificMethodFrameListener.java
@@ -41,17 +41,4 @@ public class SpecificMethodFrameListener extends BlockingMethodFrameListener
         return _expectedClass.isInstance(frame);
     }
 
-    public boolean equals(Object o)
-    {
-        if (o instanceof SpecificMethodFrameListener)
-        {
-            SpecificMethodFrameListener other = (SpecificMethodFrameListener) o;
-
-            // here we need to check if the two classes are the same.
-            return (_channelId == other._channelId) && (_expectedClass.equals(other._expectedClass));
-        }
-
-        return false;
-    }
-
 }
diff --git a/java/client/src/main/java/org/apache/qpid/client/transport/TransportConnection.java b/java/client/src/main/java/org/apache/qpid/client/transport/TransportConnection.java
index 3257caa796..e8a220f5e9 100644
--- a/java/client/src/main/java/org/apache/qpid/client/transport/TransportConnection.java
+++ b/java/client/src/main/java/org/apache/qpid/client/transport/TransportConnection.java
@@ -35,7 +35,6 @@ import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.util.HashMap;
-import java.util.Iterator;
 import java.util.Map;
 
 /**
@@ -101,27 +100,21 @@ public class TransportConnection
                             _logger.warn("Using Qpid MultiThreaded NIO - " + (System.getProperties().containsKey("qpidnio")
                                                                  ? "Qpid NIO is new default"
                                                                  : "Sysproperty 'qpidnio' is set"));
-
-
                             result = new MultiThreadSocketConnector();
                         }
                         else
                         {
                             _logger.info("Using Mina NIO");
-
                             result = new SocketConnector(); // non-blocking connector
                         }
-
                         // Don't have the connector's worker thread wait around for other connections (we only use
                         // one SocketConnector per connection at the moment anyway). This allows short-running
                         // clients (like unit tests) to complete quickly.
                         result.setWorkerTimeout(0);
-
                         return result;
                     }
                 });
                 break;
-
             case VM:
             {
                 _instance = getVMTransport(details, Boolean.getBoolean("amqj.AutoCreateVMBroker"));
@@ -280,8 +273,7 @@ public class TransportConnection
             }
 
             AMQVMBrokerCreationException amqbce =
-                    new AMQVMBrokerCreationException(null, port, because + " Stopped InVM Qpid.AMQP creation", null);
-            amqbce.initCause(e);
+                    new AMQVMBrokerCreationException(null, port, because + " Stopped InVM Qpid.AMQP creation", e);
             throw amqbce;
         }
 
@@ -294,14 +286,11 @@ public class TransportConnection
         _acceptor.unbindAll();
         synchronized (_inVmPipeAddress)
         {
-            Iterator keys = _inVmPipeAddress.keySet().iterator();
-
-            while (keys.hasNext())
-            {
-                int id = (Integer) keys.next();
-                _inVmPipeAddress.remove(id);
-            }
-        }
+            _inVmPipeAddress.clear();
+        }        
+        _acceptor = null;
+        _currentInstance = -1;
+        _currentVMPort = -1;
     }
 
     public static void killVMBroker(int port)
diff --git a/java/client/src/main/java/org/apache/qpid/client/transport/VmPipeTransportConnection.java b/java/client/src/main/java/org/apache/qpid/client/transport/VmPipeTransportConnection.java
index 25a9e26285..dca6efba67 100644
--- a/java/client/src/main/java/org/apache/qpid/client/transport/VmPipeTransportConnection.java
+++ b/java/client/src/main/java/org/apache/qpid/client/transport/VmPipeTransportConnection.java
@@ -22,15 +22,12 @@ package org.apache.qpid.client.transport;
 
 import org.apache.mina.common.ConnectFuture;
 import org.apache.mina.common.IoServiceConfig;
+import org.apache.mina.transport.vmpipe.QpidVmPipeConnector;
 import org.apache.mina.transport.vmpipe.VmPipeAddress;
 import org.apache.mina.transport.vmpipe.VmPipeConnector;
-
 import org.apache.qpid.client.protocol.AMQProtocolHandler;
 import org.apache.qpid.jms.BrokerDetails;
-import org.apache.qpid.pool.PoolingFilter;
-import org.apache.qpid.pool.ReferenceCountingExecutorService;
 import org.apache.qpid.pool.ReadWriteThreadModel;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -49,7 +46,7 @@ public class VmPipeTransportConnection implements ITransportConnection
 
     public void connect(AMQProtocolHandler protocolHandler, BrokerDetails brokerDetail) throws IOException
     {
-        final VmPipeConnector ioConnector = new VmPipeConnector();
+        final VmPipeConnector ioConnector = new QpidVmPipeConnector();
         final IoServiceConfig cfg = ioConnector.getDefaultConfig();
 
         cfg.setThreadModel(ReadWriteThreadModel.getInstance());
diff --git a/java/client/src/test/java/org/apache/qpid/client/SpecificMethodFrameListenerTest.java b/java/client/src/test/java/org/apache/qpid/client/SpecificMethodFrameListenerTest.java
deleted file mode 100644
index 69684a81ea..0000000000
--- a/java/client/src/test/java/org/apache/qpid/client/SpecificMethodFrameListenerTest.java
+++ /dev/null
@@ -1,73 +0,0 @@
-package org.apache.qpid.framing;
-
-import junit.framework.TestCase;
-import org.apache.qpid.client.state.listener.SpecificMethodFrameListener;
-import org.apache.qpid.framing.amqp_0_9.MethodRegistry_0_9;
-
-import org.apache.mina.common.ByteBuffer;
-
-/*
-*
-* 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.
-*
-*/
-
-public class SpecificMethodFrameListenerTest extends TestCase
-{
-
-    SpecificMethodFrameListener close1a = new SpecificMethodFrameListener(1, ChannelCloseOkBody.class);
-    SpecificMethodFrameListener close1b = new SpecificMethodFrameListener(1, ChannelCloseOkBody.class);
-    SpecificMethodFrameListener close2 = new SpecificMethodFrameListener(2, ChannelCloseOkBody.class);
-    SpecificMethodFrameListener open1a = new SpecificMethodFrameListener(1, ChannelOpenOkBody.class);
-    SpecificMethodFrameListener open1b = new SpecificMethodFrameListener(1, ChannelOpenOkBody.class);
-
-    public void testEquals()
-    {
-        //Check that the the same objects are equal
-        assertEquals("ChannelCloseOKBody a should equal a", close1a, close1a);
-        assertEquals("ChannelOpenOkBody a should equal a", open1a, open1a);
-
-        //check that the same values in differnt objects are equal
-        assertEquals("ChannelCloseOKBody b should equal a", close1b, close1a);
-        assertEquals("ChannelCloseOKBody a should equal b", close1a, close1b);
-        assertEquals("ChannelOpenOkBody a should equal b", open1a, open1b);
-        assertEquals("ChannelOpenOkBody a should equal b", open1a, open1b);
-
-        //Chec that different values fail
-        //Different channels
-        assertFalse("ChannelCloseOKBody channel 1 should NOT equal channel 2", close1a.equals(close2));
-        assertFalse("ChannelCloseOKBody channel 1 should NOT equal channel 2", close2.equals(close1a));
-
-        //Different Bodies
-        assertFalse("ChannelCloseOKBody should not equal ChannelOpenOkBody", close1a.equals(open1a));
-        assertFalse("ChannelOpenOkBody should not equal ChannelCloseOKBody", open1a.equals(close1a));
-    }
-
-    public void testProcessMethod() throws AMQFrameDecodingException
-    {
-        ChannelCloseOkBody ccob = MethodRegistry.getMethodRegistry(ProtocolVersion.v0_9).createChannelCloseOkBody();
-        ChannelOpenOkBody coob =  ((MethodRegistry_0_9)(MethodRegistry.getMethodRegistry(ProtocolVersion.v0_9))).createChannelOpenOkBody(new byte[0]);
-
-        assertTrue("This SpecificMethodFrameListener should process a ChannelCloseOkBody", close1a.processMethod(1, ccob));
-        assertFalse("This SpecificMethodFrameListener should NOT process a ChannelOpenOkBody", close1a.processMethod(1, coob));
-
-
-
-
-    }
-}
diff --git a/java/client/src/test/java/org/apache/qpid/test/unit/transacted/CommitRollbackTest.java b/java/client/src/test/java/org/apache/qpid/test/unit/transacted/CommitRollbackTest.java
index 224463a446..e45312448c 100644
--- a/java/client/src/test/java/org/apache/qpid/test/unit/transacted/CommitRollbackTest.java
+++ b/java/client/src/test/java/org/apache/qpid/test/unit/transacted/CommitRollbackTest.java
@@ -418,14 +418,14 @@ public class CommitRollbackTest extends TestCase
             {
                 _logger.info("Got 2 redelivered, message was prefetched");
                 _gottwoRedelivered = true;
-                
+
             }
             else
             {
-                _logger.warn("Got 2, message prefetched wasn't cleared or messages was in transit when rollback occured");                
+                _logger.warn("Got 2, message prefetched wasn't cleared or messages was in transit when rollback occured");
                 assertFalse("Already received message two", _gottwo);
                 assertFalse("Already received message redelivered two", _gottwoRedelivered);
-                
+
                 _gottwo = true;
             }
         }
@@ -437,7 +437,7 @@ public class CommitRollbackTest extends TestCase
      * This test sends two messages receives on of them but doesn't ack it.
      * The consumer is then closed
      * the first message should be returned as redelivered.
-     *  the second message should be delivered normally. 
+     *  the second message should be delivered normally.
      * @throws Exception
      */
     public void testSend2ThenCloseAfter1andTryAgain() throws Exception
@@ -458,6 +458,7 @@ public class CommitRollbackTest extends TestCase
         assertTrue("Messasge is marked as redelivered" + result, !result.getJMSRedelivered());
 
         _logger.info("Closing Consumer");
+        
         _consumer.close();
 
         _logger.info("Creating New consumer");
@@ -465,33 +466,20 @@ public class CommitRollbackTest extends TestCase
 
         _logger.info("receiving result");
 
-// NOTE: Both msg 1 & 2 will be marked as redelivered as they have both will have been rejected.
-// Only the occasion where it is not rejected will it mean it hasn't arrived at the client yet.
-        result = _consumer.receive(5000);
-        assertNotNull("test message was consumed and rolled back, but is gone", result);
 
-// The first message back will be either 1 or 2 being redelivered
-        if (result.getJMSRedelivered())
-        {
-            assertTrue("Messasge is not marked as redelivered" + result, result.getJMSRedelivered());
-        }
-        else // or it will be msg 2 arriving the first time due to latency.
-        {
-            _logger.info("Message 2 wasn't prefetched so wasn't rejected");
-            assertEquals("2", ((TextMessage) result).getText());
-        }
+        // Message 2 may be marked as redelivered if it was prefetched.
+        result = _consumer.receive(5000);
+        assertNotNull("Second message was not consumed, but is gone", result);
 
-        Message result2 = _consumer.receive(5000);
-        assertNotNull("test message was consumed and rolled back, but is gone", result2);
+        // The first message back will be 2, message 1 has been received but not committed
+        // Closing the consumer does not commit the session.
 
         // if this is message 1 then it should be marked as redelivered
-        if("1".equals(((TextMessage) result2).getText()))
+        if("1".equals(((TextMessage) result).getText()))
         {
-            assertTrue("Messasge is not marked as redelivered" + result2, result2.getJMSRedelivered());
+            fail("First message was recieved again");
         }
 
-        assertNotSame("Messages should not have the same content",((TextMessage) result2).getText(), ((TextMessage) result).getText() );
-
         result = _consumer.receive(1000);
         assertNull("test message should be null:" + result, result);