Fixed up clustering and a few tests that were failing.

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

24 files changed, 290 insertions, 257 deletions

diff --git a/java/broker/etc/log4j.xml b/java/broker/etc/log4j.xml
index 8fa6378331..133b5ce259 100644
--- a/java/broker/etc/log4j.xml
+++ b/java/broker/etc/log4j.xml
@@ -34,9 +34,9 @@
         </layout>
     </appender>
 
-    <category name="org.apache.qpid.server.queue">
+    <!--<category name="org.apache.qpid.server.queue">
         <priority value="debug"/>
-    </category>
+    </category>-->
 
     <root>
         <priority value="info"/>
diff --git a/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java b/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java
index 73264c5310..f2cd0d13cd 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java
@@ -450,11 +450,6 @@ public class AMQMessage
         _deliveredToConsumer = true;
     }
 
-    /*public void registerQueue(AMQQueue queue)
-    {
-        _destinationQueues.add(queue);
-    } */
-
     private void deliver() throws AMQException
     {
         // first we allow the handle to know that the message has been fully received. This is useful if it is
@@ -531,7 +526,7 @@ public class AMQMessage
         return buf;
     }
 
-private ByteBuffer createEncodedReturnFrame(int channelId, int replyCode, String replyText) throws AMQException
+    private ByteBuffer createEncodedReturnFrame(int channelId, int replyCode, String replyText) throws AMQException
     {
         AMQFrame returnFrame = BasicReturnBody.createAMQFrame(channelId, replyCode, replyText, getPublishBody().exchange,
                                                               getPublishBody().routingKey);
diff --git a/java/broker/src/main/log4j.properties b/java/broker/src/main/log4j.properties
deleted file mode 100644
index 3ff6f0b581..0000000000
--- a/java/broker/src/main/log4j.properties
+++ /dev/null
@@ -1,6 +0,0 @@
-log4j.rootCategory=${amqj.logging.level}, console
-
-log4j.appender.console=org.apache.log4j.ConsoleAppender
-log4j.appender.console.Threshold=info
-log4j.appender.console.layout=org.apache.log4j.PatternLayout
-log4j.appender.console.layout.ConversionPattern=%t %d %p [%c{4}] %m%n
diff --git a/java/cluster/src/main/java/org/apache/qpid/server/cluster/DefaultGroupManager.java b/java/cluster/src/main/java/org/apache/qpid/server/cluster/DefaultGroupManager.java
index 980b36cf21..4efe0b8dd0 100644
--- a/java/cluster/src/main/java/org/apache/qpid/server/cluster/DefaultGroupManager.java
+++ b/java/cluster/src/main/java/org/apache/qpid/server/cluster/DefaultGroupManager.java
@@ -19,16 +19,10 @@ package org.apache.qpid.server.cluster;
 
 import org.apache.log4j.Logger;
 import org.apache.qpid.AMQException;
-import org.apache.qpid.framing.ClusterJoinBody;
-import org.apache.qpid.framing.ClusterLeaveBody;
-import org.apache.qpid.framing.ClusterMembershipBody;
-import org.apache.qpid.framing.ClusterPingBody;
-import org.apache.qpid.framing.ClusterSuspectBody;
-import org.apache.qpid.framing.AMQMethodBody;
+import org.apache.qpid.framing.*;
 import org.apache.qpid.server.cluster.policy.StandardPolicies;
 import org.apache.qpid.server.cluster.replay.ReplayManager;
 import org.apache.qpid.server.cluster.util.LogMessage;
-import org.apache.qpid.server.cluster.util.InvokeMultiple;
 
 import java.util.List;
 
@@ -93,7 +87,7 @@ public class DefaultGroupManager implements GroupManager, MemberFailureListener,
         Broker destination = findBroker(broker);
         if(destination == null)
         {
-            _logger.warn(new LogMessage("Invalid destination sending {0}. {1} not known", message, broker));            
+            _logger.warn(new LogMessage("Invalid destination sending {0}. {1} not known", message, broker));
         }
         else
         {
@@ -114,7 +108,7 @@ public class DefaultGroupManager implements GroupManager, MemberFailureListener,
         ping.responseRequired = true;
         ping.load = _loadTable.getLocalLoad();
         BlockingHandler handler = new BlockingHandler();
-        send(getLeader(), new SimpleSendable(ping), handler);
+        send(getLeader(), new SimpleBodySendable(ping), handler);
         handler.waitForCompletion();
         if (handler.failed())
         {
@@ -157,7 +151,7 @@ public class DefaultGroupManager implements GroupManager, MemberFailureListener,
         _logger.info(new LogMessage("Connected to {0}. joining", leader));
         ClusterJoinBody join = new ClusterJoinBody();
         join.broker = _group.getLocal().getDetails();
-        send(leader, new SimpleSendable(join));
+        send(leader, new SimpleBodySendable(join));
     }
 
     private Broker connectToLeader(MemberHandle member) throws AMQException
@@ -176,7 +170,7 @@ public class DefaultGroupManager implements GroupManager, MemberFailureListener,
     {
         ClusterLeaveBody leave = new ClusterLeaveBody();
         leave.broker = _group.getLocal().getDetails();
-        send(getLeader(), new SimpleSendable(leave));
+        send(getLeader(), new SimpleBodySendable(leave));
     }
 
     private void suspect(MemberHandle broker) throws AMQException
@@ -197,7 +191,7 @@ public class DefaultGroupManager implements GroupManager, MemberFailureListener,
         {
             ClusterSuspectBody suspect = new ClusterSuspectBody();
             suspect.broker = broker.getDetails();
-            send(getLeader(), new SimpleSendable(suspect));
+            send(getLeader(), new SimpleBodySendable(suspect));
         }
     }
 
@@ -220,7 +214,7 @@ public class DefaultGroupManager implements GroupManager, MemberFailureListener,
             ClusterJoinBody request = new ClusterJoinBody();
             request.broker = member.getDetails();
             Broker leader = getLeader();
-            send(leader, new SimpleSendable(request));
+            send(leader, new SimpleBodySendable(request));
             _logger.info(new LogMessage("Passed join request for {0} to {1}", member, leader));
         }
     }
@@ -272,7 +266,7 @@ public class DefaultGroupManager implements GroupManager, MemberFailureListener,
     {
         String membership = SimpleMemberHandle.membersToString(_group.getMembers());
         ClusterMembershipBody announce = createAnnouncement(membership);
-        broadcast(new SimpleSendable(announce));
+        broadcast(new SimpleBodySendable(announce));
         _logger.info(new LogMessage("Membership announcement sent: {0}", membership));
     }
 
diff --git a/java/cluster/src/main/java/org/apache/qpid/server/cluster/SimpleBodySendable.java b/java/cluster/src/main/java/org/apache/qpid/server/cluster/SimpleBodySendable.java
new file mode 100644
index 0000000000..f7c40c60b3
--- /dev/null
+++ b/java/cluster/src/main/java/org/apache/qpid/server/cluster/SimpleBodySendable.java
@@ -0,0 +1,45 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed 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.server.cluster;
+
+import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.AMQBody;
+import org.apache.qpid.framing.AMQFrame;
+
+/**
+ */
+public class SimpleBodySendable implements Sendable
+{
+    private final AMQBody _body;
+
+    public SimpleBodySendable(AMQBody body)
+    {
+        _body = body;
+    }
+
+    public void send(int channel, Member member) throws AMQException
+    {
+        member.send(new AMQFrame(channel, _body));
+    }
+
+    public String toString()
+    {
+        return _body.toString();
+    }
+
+}
diff --git a/java/cluster/src/main/java/org/apache/qpid/server/cluster/SimpleSendable.java b/java/cluster/src/main/java/org/apache/qpid/server/cluster/SimpleSendable.java
index 4b75e76d97..f2dfaceefe 100644
--- a/java/cluster/src/main/java/org/apache/qpid/server/cluster/SimpleSendable.java
+++ b/java/cluster/src/main/java/org/apache/qpid/server/cluster/SimpleSendable.java
@@ -18,36 +18,29 @@
 package org.apache.qpid.server.cluster;
 
 import org.apache.qpid.AMQException;
-import org.apache.qpid.framing.AMQBody;
 import org.apache.qpid.framing.AMQFrame;
+import org.apache.qpid.framing.ContentBody;
+import org.apache.qpid.server.queue.AMQMessage;
 
-import java.util.Arrays;
-import java.util.List;
+import java.util.Iterator;
 
 public class SimpleSendable implements Sendable
 {
-    private final List<AMQBody> _bodies;
+    private final AMQMessage _message;
 
-    public SimpleSendable(AMQBody body)
+    public SimpleSendable(AMQMessage message)
     {
-        this(Arrays.asList(body));
-    }
-
-    public SimpleSendable(List<AMQBody> bodies)
-    {
-        _bodies = bodies;
+        _message = message;
     }
 
     public void send(int channel, Member member) throws AMQException
     {
-        for (AMQBody body : _bodies)
+        member.send(new AMQFrame(channel, _message.getPublishBody()));
+        member.send(new AMQFrame(channel, _message.getContentHeaderBody()));
+        Iterator<ContentBody> it = _message.getContentBodyIterator();
+        while (it.hasNext())
         {
-            member.send(new AMQFrame(channel, body));
+            member.send(new AMQFrame(channel, it.next()));
         }
     }
-
-    public String toString()
-    {
-        return _bodies.toString();
-    }
 }
diff --git a/java/cluster/src/main/java/org/apache/qpid/server/cluster/handler/ReplicatingHandler.java b/java/cluster/src/main/java/org/apache/qpid/server/cluster/handler/ReplicatingHandler.java
index a5fab27d16..fa1b28d49a 100644
--- a/java/cluster/src/main/java/org/apache/qpid/server/cluster/handler/ReplicatingHandler.java
+++ b/java/cluster/src/main/java/org/apache/qpid/server/cluster/handler/ReplicatingHandler.java
@@ -20,13 +20,8 @@ package org.apache.qpid.server.cluster.handler;
 import org.apache.log4j.Logger;
 import org.apache.qpid.AMQException;
 import org.apache.qpid.framing.AMQMethodBody;
-import org.apache.qpid.server.cluster.BroadcastPolicy;
-import org.apache.qpid.server.cluster.ClusteredProtocolSession;
-import org.apache.qpid.server.cluster.GroupManager;
-import org.apache.qpid.server.cluster.GroupResponseHandler;
 import org.apache.qpid.server.cluster.util.LogMessage;
-import org.apache.qpid.server.cluster.Member;
-import org.apache.qpid.server.cluster.SimpleSendable;
+import org.apache.qpid.server.cluster.*;
 import org.apache.qpid.server.cluster.policy.StandardPolicies;
 import org.apache.qpid.server.exchange.ExchangeRegistry;
 import org.apache.qpid.server.protocol.AMQMethodEvent;
@@ -78,13 +73,13 @@ class ReplicatingHandler<A extends AMQMethodBody> extends ClusterMethodHandler<A
         if (_policy == null)
         {
             //asynch delivery
-            _groupMgr.broadcast(new SimpleSendable(evt.getMethod()));
+            _groupMgr.broadcast(new SimpleBodySendable(evt.getMethod()));
             local(stateMgr, queues, exchanges, session, evt);
         }
         else
         {
             Callback callback = new Callback(stateMgr, queues, exchanges, session, evt);
-            _groupMgr.broadcast(new SimpleSendable(evt.getMethod()), _policy, callback);
+            _groupMgr.broadcast(new SimpleBodySendable(evt.getMethod()), _policy, callback);
         }
         _logger.debug(new LogMessage("Replicated {0} to peers", evt.getMethod()));
     }
diff --git a/java/cluster/src/main/java/org/apache/qpid/server/queue/ClusteredQueue.java b/java/cluster/src/main/java/org/apache/qpid/server/queue/ClusteredQueue.java
index 14d893b040..d0d6a308a7 100644
--- a/java/cluster/src/main/java/org/apache/qpid/server/queue/ClusteredQueue.java
+++ b/java/cluster/src/main/java/org/apache/qpid/server/queue/ClusteredQueue.java
@@ -58,10 +58,10 @@ public class ClusteredQueue extends AMQQueue
         _subscriptions = ((ClusteredSubscriptionManager) getSubscribers()).getAllSubscribers();
     }
 
-    public void deliver(AMQMessage message) throws AMQException
+    public void process(AMQMessage msg) throws AMQException
     {
-        _logger.info(new LogMessage("{0} delivered to clustered queue {1}", message, this));
-        super.deliver(message);
+        _logger.info(new LogMessage("{0} delivered to clustered queue {1}", msg, this));
+        super.process(msg);
     }
 
     protected void autodelete() throws AMQException
@@ -74,7 +74,7 @@ public class ClusteredQueue extends AMQQueue
             //send deletion request to all other members:
             QueueDeleteBody request = new QueueDeleteBody();
             request.queue = getName();
-            _groupMgr.broadcast(new SimpleSendable(request));
+            _groupMgr.broadcast(new SimpleBodySendable(request));
         }
     }
 
@@ -86,7 +86,7 @@ public class ClusteredQueue extends AMQQueue
         //signal other members:
         BasicCancelBody request = new BasicCancelBody();
         request.consumerTag = getName();
-        _groupMgr.broadcast(new SimpleSendable(request));
+        _groupMgr.broadcast(new SimpleBodySendable(request));
     }
 
     public void addRemoteSubcriber(MemberHandle peer)
diff --git a/java/cluster/src/main/java/org/apache/qpid/server/queue/PrivateQueue.java b/java/cluster/src/main/java/org/apache/qpid/server/queue/PrivateQueue.java
index 1e7e13a577..7940ef0778 100644
--- a/java/cluster/src/main/java/org/apache/qpid/server/queue/PrivateQueue.java
+++ b/java/cluster/src/main/java/org/apache/qpid/server/queue/PrivateQueue.java
@@ -20,6 +20,7 @@ package org.apache.qpid.server.queue;
 import org.apache.qpid.AMQException;
 import org.apache.qpid.server.cluster.SimpleSendable;
 import org.apache.qpid.server.cluster.GroupManager;
+import org.apache.qpid.server.cluster.SimpleBodySendable;
 import org.apache.qpid.framing.QueueDeleteBody;
 
 import java.util.concurrent.Executor;
@@ -55,6 +56,6 @@ public class PrivateQueue extends AMQQueue
         //send delete request to peers:
         QueueDeleteBody request = new QueueDeleteBody();
         request.queue = getName();
-        _groupMgr.broadcast(new SimpleSendable(request));
+        _groupMgr.broadcast(new SimpleBodySendable(request));
     }
 }
diff --git a/java/cluster/src/main/java/org/apache/qpid/server/queue/RemoteQueueProxy.java b/java/cluster/src/main/java/org/apache/qpid/server/queue/RemoteQueueProxy.java
index a9a467b306..2162e0146b 100644
--- a/java/cluster/src/main/java/org/apache/qpid/server/queue/RemoteQueueProxy.java
+++ b/java/cluster/src/main/java/org/apache/qpid/server/queue/RemoteQueueProxy.java
@@ -19,18 +19,13 @@ package org.apache.qpid.server.queue;
 
 import org.apache.log4j.Logger;
 import org.apache.qpid.AMQException;
-import org.apache.qpid.framing.AMQBody;
 import org.apache.qpid.framing.BasicPublishBody;
-import org.apache.qpid.framing.ContentBody;
-import org.apache.qpid.framing.ContentHeaderBody;
 import org.apache.qpid.server.cluster.ClusteredProtocolSession;
 import org.apache.qpid.server.cluster.GroupManager;
-import org.apache.qpid.server.cluster.util.LogMessage;
 import org.apache.qpid.server.cluster.MemberHandle;
 import org.apache.qpid.server.cluster.SimpleSendable;
+import org.apache.qpid.server.cluster.util.LogMessage;
 
-import java.util.ArrayList;
-import java.util.List;
 import java.util.concurrent.Executor;
 
 /**
@@ -90,17 +85,9 @@ public class RemoteQueueProxy extends AMQQueue
     void relay(AMQMessage msg) throws AMQException
     {
         BasicPublishBody publish = msg.getPublishBody();
-        ContentHeaderBody header = msg.getContentHeaderBody();
-        List<ContentBody> bodies = msg.getContentBodies();
-
-        //(i) construct a new publishing block:
-        publish.immediate = false;//can't as yet handle the immediate flag in a cluster
-        List<AMQBody> parts = new ArrayList<AMQBody>(2 + bodies.size());
-        parts.add(publish);
-        parts.add(header);
-        parts.addAll(bodies);
+        publish.immediate = false; //can't as yet handle the immediate flag in a cluster
 
-        //(ii) send this on to the broker for which it is acting as proxy:
-        _groupMgr.send(_target, new SimpleSendable(parts));
+        // send this on to the broker for which it is acting as proxy:
+        _groupMgr.send(_target, new SimpleSendable(msg));
     }
 }
diff --git a/java/cluster/src/main/java/org/apache/qpid/server/queue/RemoteSubscriptionImpl.java b/java/cluster/src/main/java/org/apache/qpid/server/queue/RemoteSubscriptionImpl.java
index 9de7a5c849..eabf374e81 100644
--- a/java/cluster/src/main/java/org/apache/qpid/server/queue/RemoteSubscriptionImpl.java
+++ b/java/cluster/src/main/java/org/apache/qpid/server/queue/RemoteSubscriptionImpl.java
@@ -49,7 +49,7 @@ class RemoteSubscriptionImpl implements Subscription, WeightedSubscriptionManage
     {
         try
         {
-            _groupMgr.send(_peer, new SimpleSendable(msg.getPayload()));
+            _groupMgr.send(_peer, new SimpleSendable(msg));
         }
         catch (AMQException e)
         {
diff --git a/java/cluster/src/test/java/org/apache/qpid/server/cluster/BrokerGroupTest.java b/java/cluster/src/test/java/org/apache/qpid/server/cluster/BrokerGroupTest.java
index 015e96f9c6..b91d7140e0 100644
--- a/java/cluster/src/test/java/org/apache/qpid/server/cluster/BrokerGroupTest.java
+++ b/java/cluster/src/test/java/org/apache/qpid/server/cluster/BrokerGroupTest.java
@@ -1,18 +1,21 @@
 /*
  *
- * Copyright (c) 2006 The Apache Software Foundation
- *
- * Licensed 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.
+ * 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.server.cluster;
diff --git a/java/cluster/src/test/java/org/apache/qpid/server/cluster/BrokerTest.java b/java/cluster/src/test/java/org/apache/qpid/server/cluster/BrokerTest.java
index d7ede7d3e0..9bee19ae95 100644
--- a/java/cluster/src/test/java/org/apache/qpid/server/cluster/BrokerTest.java
+++ b/java/cluster/src/test/java/org/apache/qpid/server/cluster/BrokerTest.java
@@ -1,18 +1,21 @@
 /*
  *
- * Copyright (c) 2006 The Apache Software Foundation
+ * 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
  *
- * Licensed 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
  *
- *    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.
+ * 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.server.cluster;
@@ -42,7 +45,7 @@ public class BrokerTest extends TestCase
                 new RecordingBroker("C", 3)
         };
         GroupResponseValidator handler = new GroupResponseValidator(new TestMethod("response"), new ArrayList<Member>(Arrays.asList(brokers)));
-        GroupRequest grpRequest = new GroupRequest(new SimpleSendable(new TestMethod("request")), StandardPolicies.SYNCH_POLICY, handler);
+        GroupRequest grpRequest = new GroupRequest(new SimpleBodySendable(new TestMethod("request")), StandardPolicies.SYNCH_POLICY, handler);
         for (Broker b : brokers)
         {
             b.invoke(grpRequest);
@@ -67,7 +70,7 @@ public class BrokerTest extends TestCase
         RecordingBroker[] succeeded = new RecordingBroker[]{a, c};
 
         GroupResponseValidator handler = new GroupResponseValidator(new TestMethod("response"), new ArrayList<Member>(Arrays.asList(succeeded)));
-        GroupRequest grpRequest = new GroupRequest(new SimpleSendable(new TestMethod("request")), StandardPolicies.SYNCH_POLICY, handler);
+        GroupRequest grpRequest = new GroupRequest(new SimpleBodySendable(new TestMethod("request")), StandardPolicies.SYNCH_POLICY, handler);
 
         for (Broker broker : all)
         {
@@ -96,7 +99,7 @@ public class BrokerTest extends TestCase
         RecordingBroker broker = new RecordingBroker("myhost", 1);
         for (AMQBody msg : msgs)
         {
-            broker.send(new SimpleSendable(msg), null);
+            broker.send(new SimpleBodySendable(msg), null);
         }
         List<AMQDataBlock> sent = broker.getMessages();
         assertEquals(msgs.length, sent.size());
@@ -112,7 +115,7 @@ public class BrokerTest extends TestCase
     {
         RecordingBroker broker = new RecordingBroker("myhost", 1);
         BlockingHandler handler = new BlockingHandler();
-        broker.send(new SimpleSendable(new TestMethod("A")), handler);
+        broker.send(new SimpleBodySendable(new TestMethod("A")), handler);
         List<AMQDataBlock> sent = broker.getMessages();
         assertEquals(1, sent.size());
         assertTrue(sent.get(0) instanceof AMQFrame);
@@ -128,7 +131,7 @@ public class BrokerTest extends TestCase
     {
         RecordingBroker broker = new RecordingBroker("myhost", 1);
         BlockingHandler handler = new BlockingHandler();
-        broker.send(new SimpleSendable(new TestMethod("A")), handler);
+        broker.send(new SimpleBodySendable(new TestMethod("A")), handler);
         List<AMQDataBlock> sent = broker.getMessages();
         assertEquals(1, sent.size());
         assertTrue(sent.get(0) instanceof AMQFrame);
diff --git a/java/cluster/src/test/java/org/apache/qpid/server/cluster/ClusterCapabilityTest.java b/java/cluster/src/test/java/org/apache/qpid/server/cluster/ClusterCapabilityTest.java
index 132ebd8ca0..76b1da8754 100644
--- a/java/cluster/src/test/java/org/apache/qpid/server/cluster/ClusterCapabilityTest.java
+++ b/java/cluster/src/test/java/org/apache/qpid/server/cluster/ClusterCapabilityTest.java
@@ -1,29 +1,30 @@
 /*
  *
- * Copyright (c) 2006 The Apache Software Foundation
- *
- * Licensed 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.
+ * 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.server.cluster;
 
-import static org.junit.Assert.assertTrue;
-import org.junit.Test;
+import junit.framework.TestCase;
 
-public class ClusterCapabilityTest
+public class ClusterCapabilityTest extends TestCase
 {
-    @Test
-    public void startWithNull()
+    public void testStartWithNull()
     {
         MemberHandle peer = new SimpleMemberHandle("myhost:9999");
         String c = ClusterCapability.add(null, peer);
@@ -31,8 +32,7 @@ public class ClusterCapabilityTest
         assertTrue(peer.matches(ClusterCapability.getPeer(c)));
     }
 
-    @Test
-    public void startWithText()
+    public void testStartWithText()
     {
         MemberHandle peer = new SimpleMemberHandle("myhost:9999");
         String c = ClusterCapability.add("existing text", peer);
diff --git a/java/cluster/src/test/java/org/apache/qpid/server/cluster/InductionBufferTest.java b/java/cluster/src/test/java/org/apache/qpid/server/cluster/InductionBufferTest.java
index fedf47d49a..7e58add91e 100644
--- a/java/cluster/src/test/java/org/apache/qpid/server/cluster/InductionBufferTest.java
+++ b/java/cluster/src/test/java/org/apache/qpid/server/cluster/InductionBufferTest.java
@@ -1,18 +1,21 @@
 /*
  *
- * Copyright (c) 2006 The Apache Software Foundation
- *
- * Licensed 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.
+ * 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.server.cluster;
diff --git a/java/cluster/src/test/java/org/apache/qpid/server/cluster/RecordingBroker.java b/java/cluster/src/test/java/org/apache/qpid/server/cluster/RecordingBroker.java
index 388d584288..1ec5154a98 100644
--- a/java/cluster/src/test/java/org/apache/qpid/server/cluster/RecordingBroker.java
+++ b/java/cluster/src/test/java/org/apache/qpid/server/cluster/RecordingBroker.java
@@ -1,18 +1,21 @@
 /*
  *
- * Copyright (c) 2006 The Apache Software Foundation
- *
- * Licensed 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.
+ * 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.server.cluster;
diff --git a/java/cluster/src/test/java/org/apache/qpid/server/cluster/RecordingBrokerFactory.java b/java/cluster/src/test/java/org/apache/qpid/server/cluster/RecordingBrokerFactory.java
index e5e95323af..d3e972e273 100644
--- a/java/cluster/src/test/java/org/apache/qpid/server/cluster/RecordingBrokerFactory.java
+++ b/java/cluster/src/test/java/org/apache/qpid/server/cluster/RecordingBrokerFactory.java
@@ -1,18 +1,21 @@
 /*
  *
- * Copyright (c) 2006 The Apache Software Foundation
- *
- * Licensed 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.
+ * 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.server.cluster;
diff --git a/java/cluster/src/test/java/org/apache/qpid/server/cluster/SimpleClusterTest.java b/java/cluster/src/test/java/org/apache/qpid/server/cluster/SimpleClusterTest.java
index a4d13ea46d..c427285f4a 100644
--- a/java/cluster/src/test/java/org/apache/qpid/server/cluster/SimpleClusterTest.java
+++ b/java/cluster/src/test/java/org/apache/qpid/server/cluster/SimpleClusterTest.java
@@ -1,23 +1,25 @@
 /*
  *
- * Copyright (c) 2006 The Apache Software Foundation
- *
- * Licensed 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.
+ * 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.server.cluster;
 
-import org.junit.Test;
 import org.apache.qpid.AMQException;
 import org.apache.qpid.url.URLSyntaxException;
 import org.apache.qpid.client.AMQConnection;
@@ -25,10 +27,11 @@ import org.apache.qpid.client.AMQSession;
 
 import javax.jms.JMSException;
 
-public class SimpleClusterTest
+import junit.framework.TestCase;
+
+public class SimpleClusterTest extends TestCase
 {
-    @Test
-    public void declareExchange() throws AMQException, JMSException, URLSyntaxException
+    public void testDeclareExchange() throws AMQException, JMSException, URLSyntaxException
     {
         AMQConnection con = new AMQConnection("localhost:9000", "guest", "guest", "test", "/test");
         AMQSession session = (AMQSession) con.createSession(false, AMQSession.NO_ACKNOWLEDGE);
diff --git a/java/cluster/src/test/java/org/apache/qpid/server/cluster/SimpleMemberHandleTest.java b/java/cluster/src/test/java/org/apache/qpid/server/cluster/SimpleMemberHandleTest.java
index f7c728759b..8ff8357377 100644
--- a/java/cluster/src/test/java/org/apache/qpid/server/cluster/SimpleMemberHandleTest.java
+++ b/java/cluster/src/test/java/org/apache/qpid/server/cluster/SimpleMemberHandleTest.java
@@ -1,39 +1,37 @@
 /*
  *
- * Copyright (c) 2006 The Apache Software Foundation
- *
- * Licensed 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.
+ * 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.server.cluster;
 
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
-import org.junit.Test;
+import junit.framework.TestCase;
 
-public class SimpleMemberHandleTest
+public class SimpleMemberHandleTest extends TestCase
 {
-    @Test
-    public void matches()
+    public void testMatches()
     {
         assertMatch(new SimpleMemberHandle("localhost", 8888), new SimpleMemberHandle("localhost", 8888));
         assertNoMatch(new SimpleMemberHandle("localhost", 8889), new SimpleMemberHandle("localhost", 8888));
         assertNoMatch(new SimpleMemberHandle("localhost", 8888), new SimpleMemberHandle("localhost2", 8888));
     }
 
-
-    @Test
-    public void resolve()
+    public void testResolve()
     {
         assertEquivalent(new SimpleMemberHandle("WGLAIBD8XGR0J:9000"), new SimpleMemberHandle("localhost:9000"));
     }
diff --git a/java/cluster/src/test/java/org/apache/qpid/server/cluster/TestBroker.java b/java/cluster/src/test/java/org/apache/qpid/server/cluster/TestBroker.java
index c4a1985ae3..d3ccbf0ac6 100644
--- a/java/cluster/src/test/java/org/apache/qpid/server/cluster/TestBroker.java
+++ b/java/cluster/src/test/java/org/apache/qpid/server/cluster/TestBroker.java
@@ -1,18 +1,21 @@
 /*
  *
- * Copyright (c) 2006 The Apache Software Foundation
- *
- * Licensed 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.
+ * 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.server.cluster;
diff --git a/java/cluster/src/test/java/org/apache/qpid/server/cluster/TestBrokerFactory.java b/java/cluster/src/test/java/org/apache/qpid/server/cluster/TestBrokerFactory.java
index cd4e340925..92eaec876a 100644
--- a/java/cluster/src/test/java/org/apache/qpid/server/cluster/TestBrokerFactory.java
+++ b/java/cluster/src/test/java/org/apache/qpid/server/cluster/TestBrokerFactory.java
@@ -1,18 +1,21 @@
 /*
  *
- * Copyright (c) 2006 The Apache Software Foundation
- *
- * Licensed 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.
+ * 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.server.cluster;
diff --git a/java/cluster/src/test/java/org/apache/qpid/server/cluster/TestReplayManager.java b/java/cluster/src/test/java/org/apache/qpid/server/cluster/TestReplayManager.java
index e2d6f75f19..c529c83cc0 100644
--- a/java/cluster/src/test/java/org/apache/qpid/server/cluster/TestReplayManager.java
+++ b/java/cluster/src/test/java/org/apache/qpid/server/cluster/TestReplayManager.java
@@ -1,18 +1,21 @@
 /*
  *
- * Copyright (c) 2006 The Apache Software Foundation
- *
- * Licensed 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.
+ * 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.server.cluster;
diff --git a/java/cluster/src/test/java/org/apache/qpid/server/cluster/TestSession.java b/java/cluster/src/test/java/org/apache/qpid/server/cluster/TestSession.java
index 675e20c9dc..86ec808924 100644
--- a/java/cluster/src/test/java/org/apache/qpid/server/cluster/TestSession.java
+++ b/java/cluster/src/test/java/org/apache/qpid/server/cluster/TestSession.java
@@ -1,18 +1,21 @@
 /*
  *
- * Copyright (c) 2006 The Apache Software Foundation
- *
- * Licensed 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.
+ * 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.server.cluster;
diff --git a/java/systests/src/test/java/org/apache/qpid/test/unit/ack/DisconnectAndRedeliverTest.java b/java/systests/src/test/java/org/apache/qpid/test/unit/ack/DisconnectAndRedeliverTest.java
index 5de91ac80c..3e35e3c85b 100644
--- a/java/systests/src/test/java/org/apache/qpid/test/unit/ack/DisconnectAndRedeliverTest.java
+++ b/java/systests/src/test/java/org/apache/qpid/test/unit/ack/DisconnectAndRedeliverTest.java
@@ -59,6 +59,7 @@ public class DisconnectAndRedeliverTest extends TestCase
     protected void tearDown() throws Exception
     {
         super.tearDown();
+        TransportConnection.killAllVMBrokers();
     }
 
     /**