Index: src/org/jgroups/protocols/TP.java
===================================================================
RCS file: /cvsroot/javagroups/JGroups/src/org/jgroups/protocols/TP.java,v
retrieving revision 1.160.2.15
diff -u -r1.160.2.15 TP.java
--- src/org/jgroups/protocols/TP.java	23 Apr 2008 09:23:48 -0000	1.160.2.15
+++ src/org/jgroups/protocols/TP.java	6 May 2008 19:46:16 -0000
@@ -1499,8 +1499,8 @@
     protected ExecutorService createThreadPool(int min_threads, int max_threads, long keep_alive_time, String rejection_policy,
                                                BlockingQueue<Runnable> queue, final String thread_name) {
 
-        ThreadPoolExecutor pool=new ThreadPoolExecutor(min_threads, max_threads, keep_alive_time, TimeUnit.MILLISECONDS, queue);
-        pool.setThreadFactory(ProtocolStack.newIDThreadFactory(thread_naming_pattern, pool_thread_group, thread_name, false));
+        TCLActionThreadPoolExecutor pool=new TCLActionThreadPoolExecutor(min_threads, max_threads, keep_alive_time, TimeUnit.MILLISECONDS, queue, true);
+        pool.setThreadFactory(ProtocolStack.newIDThreadFactory(thread_naming_pattern, pool_thread_group, thread_name, false, true));
 
         //default
         RejectedExecutionHandler handler = new ThreadPoolExecutor.CallerRunsPolicy();
Index: src/org/jgroups/stack/ProtocolStack.java
===================================================================
RCS file: /cvsroot/javagroups/JGroups/src/org/jgroups/stack/ProtocolStack.java,v
retrieving revision 1.59.2.5
diff -u -r1.59.2.5 ProtocolStack.java
--- src/org/jgroups/stack/ProtocolStack.java	3 Apr 2008 14:58:01 -0000	1.59.2.5
+++ src/org/jgroups/stack/ProtocolStack.java	6 May 2008 19:46:16 -0000
@@ -53,7 +53,7 @@
 
 
     public ProtocolStack(JChannel channel, String setup_string) throws ChannelException {
-        this(newThreadFactory(Util.getGlobalThreadGroup(),"",false),channel,setup_string);
+        this(newThreadFactory(Util.getGlobalThreadGroup(),"",false, false),channel,setup_string);
     }
     
     public ProtocolStack(ThreadFactory factory, JChannel channel, String setup_string) throws ChannelException {
@@ -62,7 +62,7 @@
         this.channel=channel;
         ClassConfigurator.getInstance(true); // will create the singleton
         this.timer_thread_factory=new IdThreadFactory(
-                newThreadFactory(new ThreadGroup(Util.getGlobalThreadGroup(), "Timers"), "Timer", true),
+                newThreadFactory(new ThreadGroup(Util.getGlobalThreadGroup(), "Timers"), "Timer", true, false),
                 null);
         timer= new TimeScheduler(timer_thread_factory);
     }
@@ -418,16 +418,16 @@
         return null;
     }
     
-    public static ThreadFactory newThreadFactory(ThreadGroup group,String baseName, boolean createDaemons){
-        return new DefaultThreadFactory(group,baseName, createDaemons);
+    public static ThreadFactory newThreadFactory(ThreadGroup group,String baseName, boolean createDaemons, boolean useTCCL){
+        return new DefaultThreadFactory(group,baseName, createDaemons, useTCCL);
     }
 
-    public static ThreadFactory newThreadFactory(ThreadNamingPattern pattern,ThreadGroup group,String baseName, boolean createDaemons){
-        return new PatternedThreadFactory(new DefaultThreadFactory(group,baseName, createDaemons),pattern);
+    public static ThreadFactory newThreadFactory(ThreadNamingPattern pattern,ThreadGroup group,String baseName, boolean createDaemons, boolean useTCCL){
+        return new PatternedThreadFactory(new DefaultThreadFactory(group,baseName, createDaemons, useTCCL),pattern);
     }
     
-    public static ThreadFactory newIDThreadFactory(ThreadNamingPattern pattern,ThreadGroup group,String baseName, boolean createDaemons) {
-        return new IdThreadFactory(new DefaultThreadFactory(group,baseName, createDaemons),pattern);
+    public static ThreadFactory newIDThreadFactory(ThreadNamingPattern pattern,ThreadGroup group,String baseName, boolean createDaemons, boolean useTCCL) {
+        return new IdThreadFactory(new DefaultThreadFactory(group,baseName, createDaemons, useTCCL),pattern);
     }
 
 
@@ -436,11 +436,14 @@
         private final ThreadGroup group;
         private final String baseName;
         private final boolean createDaemons;   
+        private final TCLAction tclAction;
         
-        public DefaultThreadFactory(ThreadGroup group,String baseName, boolean createDaemons){
+        public DefaultThreadFactory(ThreadGroup group,String baseName, boolean createDaemons, boolean useTCCL){
             this.group = group;
             this.baseName = baseName;
-            this.createDaemons = createDaemons;               
+            this.createDaemons = createDaemons;          
+            this.tclAction = useTCCL ? TCLAction.FACTORY.getTCLAction() : 
+                                       TCLAction.FACTORY.getTCLAction(DefaultThreadFactory.class);
         }
               
         public Thread newThread(Runnable r, String name) {
@@ -454,6 +457,8 @@
         public Thread newThread(ThreadGroup group, Runnable r, String name) {
             Thread thread = new Thread(group, r, name);
             thread.setDaemon(createDaemons);
+            if (tclAction != null)
+               tclAction.setContextClassLoader(thread);
             return thread;
         }
     }
Index: src/org/jgroups/util/TimeScheduler.java
===================================================================
RCS file: /cvsroot/javagroups/JGroups/src/org/jgroups/util/TimeScheduler.java,v
retrieving revision 1.23.4.1
diff -u -r1.23.4.1 TimeScheduler.java
--- src/org/jgroups/util/TimeScheduler.java	21 Apr 2008 08:44:09 -0000	1.23.4.1
+++ src/org/jgroups/util/TimeScheduler.java	6 May 2008 19:46:16 -0000
@@ -71,9 +71,8 @@
             log.error("could not set number of timer threads", e);
         }
     }
-
-
-
+    
+    private TCLAction tclAction;
 
     /**
      * Create a scheduler that executes tasks in dynamically adjustable intervals
@@ -83,8 +82,14 @@
     }
 
     public TimeScheduler(ThreadFactory factory) {
+        this(factory, TCLAction.FACTORY.getTCLAction(TimeScheduler.class));
+        setRejectedExecutionHandler(new ShutdownRejectedExecutionHandler(getRejectedExecutionHandler()));
+    }
+
+    public TimeScheduler(ThreadFactory factory, TCLAction tclAction) {
         super(TIMER_DEFAULT_NUM_THREADS, factory);
         setRejectedExecutionHandler(new ShutdownRejectedExecutionHandler(getRejectedExecutionHandler()));
+        this.tclAction = tclAction;
     }
 
     public TimeScheduler(int corePoolSize) {
@@ -143,8 +148,7 @@
         return getQueue().size();
     }
 
-
-    /**
+   /**
      * Start the scheduler, if it's suspended or stopped
      */
     public void start() {
@@ -170,6 +174,18 @@
         getQueue().clear();
         awaitTermination(Global.THREADPOOL_SHUTDOWN_WAIT_TIME, TimeUnit.MILLISECONDS);
     }
+    
+    /**
+     * Overrides the superclass method to restore this class' classloader
+     * as the current thread's context classloader.
+     */
+    protected void afterExecute(Runnable r, Throwable t)
+    {
+        super.afterExecute(r, t);
+        
+        if (tclAction != null)
+            tclAction.setContextClassLoader();        
+    }
 
 
     private class TaskWrapper<V> implements Runnable, ScheduledFuture<V> {
Index: src/org/jgroups/mux/Multiplexer.java
===================================================================
RCS file: /cvsroot/javagroups/JGroups/src/org/jgroups/mux/Multiplexer.java,v
retrieving revision 1.85.2.15
diff -u -r1.85.2.15 Multiplexer.java
--- src/org/jgroups/mux/Multiplexer.java	21 Apr 2008 08:44:10 -0000	1.85.2.15
+++ src/org/jgroups/mux/Multiplexer.java	6 May 2008 19:46:16 -0000
@@ -217,11 +217,12 @@
         max_threads=Global.getPropertyAsInteger(Global.MUX_MAX_THREADS, max_threads);
         keep_alive=Global.getPropertyAsLong(Global.MUX_KEEPALIVE, keep_alive);
 
-        org.jgroups.util.ThreadFactory factory = ProtocolStack.newThreadFactory(pattern, new ThreadGroup(Util.getGlobalThreadGroup(), "MultiplexerThreads"), "Multiplexer", false);
-        return new ThreadPoolExecutor(min_threads, max_threads, keep_alive, TimeUnit.MILLISECONDS,
-                                      new SynchronousQueue<Runnable>(), 
-                                      factory,
-                                      new ShutdownRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()));
+        org.jgroups.util.ThreadFactory factory = ProtocolStack.newThreadFactory(pattern, new ThreadGroup(Util.getGlobalThreadGroup(), "MultiplexerThreads"), "Multiplexer", false, true);
+        return new TCLActionThreadPoolExecutor(min_threads, max_threads, keep_alive, TimeUnit.MILLISECONDS,
+                                               new SynchronousQueue<Runnable>(), 
+                                               factory,
+                                               new ShutdownRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()),
+                                               true);
     }
 
     protected void shutdownThreadPool() {
Index: conf/classloader-leak-test.xml
===================================================================
RCS file: conf/classloader-leak-test.xml
diff -N conf/classloader-leak-test.xml
--- /dev/null	1 Jan 1970 00:00:00 -0000
+++ conf/classloader-leak-test.xml	1 Jan 1970 00:00:00 -0000
@@ -0,0 +1,411 @@
+<!--
+  JGroups protocol stacks definitions used by the JChannelFactory unit tests.
+  Version: $Id:jgroups-channelfactory-stacks.xml 71313 2008-03-26 19:46:59Z bstansberry@jboss.com $
+-->
+<protocol_stacks>
+
+    <stack name="shared1"
+           description="shared stack1">
+        <config>
+          <UDP
+             singleton_name="shared1"
+             mcast_port="${jgroups.udp.mcast_port:35688}"
+             mcast_addr="${jgroups.udp.mcast_addr:228.111.111.111}"
+             tos="8"
+             ucast_recv_buf_size="20000000"
+             ucast_send_buf_size="640000"
+             mcast_recv_buf_size="25000000"
+             mcast_send_buf_size="640000"
+             loopback="true"
+             discard_incompatible_packets="true"
+             max_bundle_size="64000"
+             max_bundle_timeout="30"
+             use_incoming_packet_handler="true"
+             ip_ttl="${jgroups.udp.ip_ttl:2}"
+             enable_bundling="false"
+                 
+             use_concurrent_stack="true"
+
+		     thread_pool.enabled="true"
+		     thread_pool.min_threads="1"
+		     thread_pool.max_threads="200"
+		     thread_pool.keep_alive_time="5000"
+           thread_pool.queue_enabled="false"
+           thread_pool.queue_max_size="1000"
+		     thread_pool.rejection_policy="Run"
+		
+		     oob_thread_pool.enabled="true"
+		     oob_thread_pool.min_threads="1"
+		     oob_thread_pool.max_threads="8"
+		     oob_thread_pool.keep_alive_time="5000"
+		     oob_thread_pool.queue_enabled="false"
+		     oob_thread_pool.queue_max_size="100"
+		     oob_thread_pool.rejection_policy="Run"/>
+          <PING timeout="1000" num_initial_members="1"/>
+          <MERGE2 max_interval="100000" min_interval="20000"/>
+          <FD_SOCK/>
+          <FD timeout="10000" max_tries="5" shun="true"/>
+          <VERIFY_SUSPECT timeout="1500"/>
+          <pbcast.NAKACK use_mcast_xmit="false" gc_lag="0"
+                   retransmit_timeout="300,600,1200,2400,4800"
+                   discard_delivered_msgs="true"/>
+          <UNICAST timeout="300,600,1200,2400,3600"/>
+          <pbcast.STABLE stability_delay="1000" desired_avg_gossip="50000"
+                   max_bytes="400000"/>
+          <pbcast.GMS print_local_addr="true" join_timeout="3000"
+                   shun="true"
+                   view_bundling="true"
+                   view_ack_collection_timeout="5000"/>
+          <FC max_credits="2000000" min_threshold="0.10"/>
+          <FRAG2 frag_size="60000"/>
+          <!-- pbcast.STREAMING_STATE_TRANSFER use_reading_thread="true"/ -->
+          <pbcast.STATE_TRANSFER/>
+          <pbcast.FLUSH timeout="0"/>
+        </config>
+    </stack>
+    
+    <stack name="shared2"
+           description="shared stack2">
+        <config>
+          <UDP
+             singleton_name="shared2"
+             mcast_port="${jgroups.udp.mcast_port:35688}"
+             mcast_addr="${jgroups.udp.mcast_addr:228.111.111.111}"
+             tos="8"
+             ucast_recv_buf_size="20000000"
+             ucast_send_buf_size="640000"
+             mcast_recv_buf_size="25000000"
+             mcast_send_buf_size="640000"
+             loopback="true"
+             discard_incompatible_packets="true"
+             max_bundle_size="64000"
+             max_bundle_timeout="30"
+             use_incoming_packet_handler="true"
+             ip_ttl="${jgroups.udp.ip_ttl:2}"
+             enable_bundling="false"
+                 
+             use_concurrent_stack="true"
+
+           thread_pool.enabled="true"
+           thread_pool.min_threads="1"
+           thread_pool.max_threads="200"
+           thread_pool.keep_alive_time="5000"
+           thread_pool.queue_enabled="false"
+           thread_pool.queue_max_size="1000"
+           thread_pool.rejection_policy="Run"
+      
+           oob_thread_pool.enabled="true"
+           oob_thread_pool.min_threads="1"
+           oob_thread_pool.max_threads="8"
+           oob_thread_pool.keep_alive_time="5000"
+           oob_thread_pool.queue_enabled="false"
+           oob_thread_pool.queue_max_size="100"
+           oob_thread_pool.rejection_policy="Run"/>
+          <PING timeout="1000" num_initial_members="1"/>
+          <MERGE2 max_interval="100000" min_interval="20000"/>
+          <FD_SOCK/>
+          <FD timeout="10000" max_tries="5" shun="true"/>
+          <VERIFY_SUSPECT timeout="1500"/>
+          <pbcast.NAKACK use_mcast_xmit="false" gc_lag="0"
+                   retransmit_timeout="300,600,1200,2400,4800"
+                   discard_delivered_msgs="true"/>
+          <UNICAST timeout="300,600,1200,2400,3600"/>
+          <pbcast.STABLE stability_delay="1000" desired_avg_gossip="50000"
+                   max_bytes="400000"/>
+          <pbcast.GMS print_local_addr="true" join_timeout="3000"
+                   shun="true"
+                   view_bundling="true"
+                   view_ack_collection_timeout="5000"/>
+          <FC max_credits="2000000" min_threshold="0.10"/>
+          <FRAG2 frag_size="60000"/>
+          <!-- pbcast.STREAMING_STATE_TRANSFER use_reading_thread="true"/ -->
+          <pbcast.STATE_TRANSFER/>
+          <pbcast.FLUSH timeout="0"/>
+        </config>
+    </stack>
+    
+    <stack name="unshared"
+           description="unshared stack">
+        <config>
+          <UDP
+             mcast_port="${jgroups.udp.mcast_port:35689}"
+             mcast_addr="${jgroups.udp.mcast_addr:228.112.112.112}"
+             tos="8"
+             ucast_recv_buf_size="20000000"
+             ucast_send_buf_size="640000"
+             mcast_recv_buf_size="25000000"
+             mcast_send_buf_size="640000"
+             loopback="true"
+             discard_incompatible_packets="true"
+             max_bundle_size="64000"
+             max_bundle_timeout="30"
+             use_incoming_packet_handler="true"
+             ip_ttl="${jgroups.udp.ip_ttl:2}"
+             enable_bundling="false"
+                 
+             use_concurrent_stack="true"
+
+           thread_pool.enabled="true"
+           thread_pool.min_threads="1"
+           thread_pool.max_threads="200"
+           thread_pool.keep_alive_time="5000"
+           thread_pool.queue_enabled="false"
+           thread_pool.queue_max_size="1000"
+           thread_pool.rejection_policy="Run"
+      
+           oob_thread_pool.enabled="true"
+           oob_thread_pool.min_threads="1"
+           oob_thread_pool.max_threads="8"
+           oob_thread_pool.keep_alive_time="5000"
+           oob_thread_pool.queue_enabled="false"
+           oob_thread_pool.queue_max_size="100"
+           oob_thread_pool.rejection_policy="Run"/>
+          <PING timeout="1000" num_initial_members="1"/>
+          <MERGE2 max_interval="100000" min_interval="20000"/>
+          <FD_SOCK/>
+          <FD timeout="10000" max_tries="5" shun="true"/>
+          <VERIFY_SUSPECT timeout="1500"/>
+          <pbcast.NAKACK use_mcast_xmit="false" gc_lag="0"
+                   retransmit_timeout="300,600,1200,2400,4800"
+                   discard_delivered_msgs="true"/>
+          <UNICAST timeout="300,600,1200,2400,3600"/>
+          <pbcast.STABLE stability_delay="1000" desired_avg_gossip="50000"
+                   max_bytes="400000"/>
+          <pbcast.GMS print_local_addr="true" join_timeout="3000"
+                   shun="true"
+                   view_bundling="true"
+                   view_ack_collection_timeout="5000"/>
+          <FC max_credits="2000000" min_threshold="0.10"/>
+          <FRAG2 frag_size="60000"/>
+          <!-- pbcast.STREAMING_STATE_TRANSFER use_reading_thread="true"/ -->
+          <pbcast.STATE_TRANSFER/>
+          <pbcast.FLUSH timeout="0"/>
+        </config>
+    </stack>
+
+
+    <stack name="tcp"
+           description="TCP based stack, with flow control and message bundling. This is usually used when IP
+           multicasting cannot be used in a network, e.g. because it is disabled (routers discard multicast).
+           Note that TCP.bind_addr and TCPPING.initial_hosts should be set, possibly via system properties, e.g.
+           -Djgroups.bind_addr=192.168.5.2 and -Djgroups.tcpping.initial_hosts=192.168.5.2[7800]">
+        <config>
+            <TCP start_port="7800"
+                 loopback="true"
+                 recv_buf_size="20000000"
+                 send_buf_size="640000"
+                 discard_incompatible_packets="true"
+                 max_bundle_size="64000"
+                 max_bundle_timeout="30"
+                 use_incoming_packet_handler="true"
+                 enable_bundling="true"
+                 use_send_queues="false"
+                 sock_conn_timeout="300"
+                 skip_suspected_members="true"
+                 
+                 use_concurrent_stack="true"
+   
+               thread_pool.enabled="true"
+               thread_pool.min_threads="1"
+               thread_pool.max_threads="25"
+               thread_pool.keep_alive_time="5000"
+               thread_pool.queue_enabled="false"
+               thread_pool.queue_max_size="100"
+               thread_pool.rejection_policy="run"
+      
+               oob_thread_pool.enabled="true"
+               oob_thread_pool.min_threads="1"
+               oob_thread_pool.max_threads="8"
+               oob_thread_pool.keep_alive_time="5000"
+               oob_thread_pool.queue_enabled="false"
+               oob_thread_pool.queue_max_size="100"
+               oob_thread_pool.rejection_policy="run"/>
+               
+            <TCPPING timeout="3000"
+                     initial_hosts="${jgroups.tcpping.initial_hosts:localhost[7800],localhost[7801]}"
+                     port_range="1"
+                     num_initial_members="3"/>
+            <MERGE2 max_interval="100000"
+                      min_interval="20000"/>
+            <FD_SOCK/>
+            <FD timeout="10000" max_tries="5"   shun="true"/>
+            <VERIFY_SUSPECT timeout="1500"  />
+            <pbcast.NAKACK
+                           use_mcast_xmit="false" gc_lag="0"
+                           retransmit_timeout="300,600,1200,2400,4800"
+                           discard_delivered_msgs="false"/>
+            <UNICAST timeout="300,600,1200"/>
+            <pbcast.STABLE stability_delay="1000" desired_avg_gossip="50000"
+                           max_bytes="400000"/>
+            <pbcast.GMS print_local_addr="true" join_timeout="5000"  
+                        shun="false"
+                        view_bundling="true"/>
+            <FC max_credits="2000000"
+                min_threshold="0.10"/>
+            <FRAG2 frag_size="60000"  />
+            <pbcast.STREAMING_STATE_TRANSFER/>
+           <!-- <pbcast.STATE_TRANSFER/> -->
+           <pbcast.FLUSH timeout="0"/>
+        </config>
+    </stack>
+    
+    <!--  Here we don't use the concurrent stack, so an incoming packet handler 
+          thread carries message -->
+    <stack name="nonconcurrent1"
+           description="no concurrent stack1">
+        <config>
+          <UDP
+             singleton_name="nonconcurrent1"
+             mcast_port="${jgroups.udp.mcast_port:35799}"
+             mcast_addr="${jgroups.udp.mcast_addr:228.114.114.114}"
+             tos="8"
+             ucast_recv_buf_size="20000000"
+             ucast_send_buf_size="640000"
+             mcast_recv_buf_size="25000000"
+             mcast_send_buf_size="640000"
+             loopback="true"
+             discard_incompatible_packets="true"
+             max_bundle_size="64000"
+             max_bundle_timeout="30"
+             use_incoming_packet_handler="true"
+             ip_ttl="${jgroups.udp.ip_ttl:2}"
+             enable_bundling="false"
+                 
+             use_concurrent_stack="false"
+
+             thread_pool.enabled="false"
+      
+             oob_thread_pool.enabled="false"/>  
+          <PING timeout="1000" num_initial_members="1"/>
+          <MERGE2 max_interval="100000" min_interval="20000"/>
+          <FD_SOCK/>
+          <FD timeout="10000" max_tries="5" shun="true"/>
+          <VERIFY_SUSPECT timeout="1500"/>
+          <pbcast.NAKACK use_mcast_xmit="false" gc_lag="0"
+                   retransmit_timeout="300,600,1200,2400,4800"
+                   discard_delivered_msgs="true"/>
+          <UNICAST timeout="300,600,1200,2400,3600"/>
+          <pbcast.STABLE stability_delay="1000" desired_avg_gossip="50000"
+                   max_bytes="400000"/>
+          <pbcast.GMS print_local_addr="true" join_timeout="3000"
+                   shun="true"
+                   view_bundling="true"
+                   view_ack_collection_timeout="5000"/>
+          <FC max_credits="2000000" min_threshold="0.10"/>
+          <FRAG2 frag_size="60000"/>
+          <!-- pbcast.STREAMING_STATE_TRANSFER use_reading_thread="true"/ -->
+          <pbcast.STATE_TRANSFER/>
+          <pbcast.FLUSH timeout="0"/>
+        </config>
+    </stack>
+    
+    <stack name="nonconcurrent2"
+           description="no concurrent stack2">
+        <config>
+          <UDP
+             singleton_name="nonconcurrent2"
+             mcast_port="${jgroups.udp.mcast_port:35799}"
+             mcast_addr="${jgroups.udp.mcast_addr:228.114.114.114}"
+             tos="8"
+             ucast_recv_buf_size="20000000"
+             ucast_send_buf_size="640000"
+             mcast_recv_buf_size="25000000"
+             mcast_send_buf_size="640000"
+             loopback="true"
+             discard_incompatible_packets="true"
+             max_bundle_size="64000"
+             max_bundle_timeout="30"
+             use_incoming_packet_handler="true"
+             ip_ttl="${jgroups.udp.ip_ttl:2}"
+             enable_bundling="false"
+                 
+             use_concurrent_stack="false"
+
+             thread_pool.enabled="false"
+      
+             oob_thread_pool.enabled="false"/>  
+          <PING timeout="1000" num_initial_members="1"/>
+          <MERGE2 max_interval="100000" min_interval="20000"/>
+          <FD_SOCK/>
+          <FD timeout="10000" max_tries="5" shun="true"/>
+          <VERIFY_SUSPECT timeout="1500"/>
+          <pbcast.NAKACK use_mcast_xmit="false" gc_lag="0"
+                   retransmit_timeout="300,600,1200,2400,4800"
+                   discard_delivered_msgs="true"/>
+          <UNICAST timeout="300,600,1200,2400,3600"/>
+          <pbcast.STABLE stability_delay="1000" desired_avg_gossip="50000"
+                   max_bytes="400000"/>
+          <pbcast.GMS print_local_addr="true" join_timeout="3000"
+                   shun="true"
+                   view_bundling="true"
+                   view_ack_collection_timeout="5000"/>
+          <FC max_credits="2000000" min_threshold="0.10"/>
+          <FRAG2 frag_size="60000"/>
+          <!-- pbcast.STREAMING_STATE_TRANSFER use_reading_thread="true"/ -->
+          <pbcast.STATE_TRANSFER/>
+          <pbcast.FLUSH timeout="0"/>
+        </config>
+    </stack>
+
+    <stack name="queues" description="to check queue configuration">
+        <config>
+          <UDP
+             singleton_name="queues"
+             mcast_port="${jgroups.udp.mcast_port:35688}"
+             mcast_addr="${jgroups.udp.mcast_addr:228.111.111.111}"
+             tos="8"
+             ucast_recv_buf_size="20000000"
+             ucast_send_buf_size="640000"
+             mcast_recv_buf_size="25000000"
+             mcast_send_buf_size="640000"
+             loopback="true"
+             discard_incompatible_packets="true"
+             max_bundle_size="64000"
+             max_bundle_timeout="30"
+             use_incoming_packet_handler="true"
+             ip_ttl="${jgroups.udp.ip_ttl:2}"
+             enable_bundling="false"
+                 
+             use_concurrent_stack="true"
+
+           thread_pool.enabled="true"
+           thread_pool.min_threads="2"
+           thread_pool.max_threads="22"
+           thread_pool.keep_alive_time="3000"
+           thread_pool.queue_enabled="true"
+           thread_pool.queue_max_size="750"
+           thread_pool.rejection_policy="Discard"
+      
+           oob_thread_pool.enabled="true"
+           oob_thread_pool.min_threads="3"
+           oob_thread_pool.max_threads="12"
+           oob_thread_pool.keep_alive_time="4000"
+           oob_thread_pool.queue_enabled="true"
+           oob_thread_pool.queue_max_size="75"
+           oob_thread_pool.rejection_policy="Abort"/>
+          <PING timeout="1000" num_initial_members="1"/>
+          <MERGE2 max_interval="100000" min_interval="20000"/>
+          <FD_SOCK/>
+          <FD timeout="10000" max_tries="5" shun="true"/>
+          <VERIFY_SUSPECT timeout="1500"/>
+          <pbcast.NAKACK use_mcast_xmit="false" gc_lag="0"
+                   retransmit_timeout="300,600,1200,2400,4800"
+                   discard_delivered_msgs="true"/>
+          <UNICAST timeout="300,600,1200,2400,3600"/>
+          <pbcast.STABLE stability_delay="1000" desired_avg_gossip="50000"
+                   max_bytes="400000"/>
+          <pbcast.GMS print_local_addr="true" join_timeout="3000"
+                   shun="true"
+                   view_bundling="true"
+                   view_ack_collection_timeout="5000"/>
+          <FC max_credits="2000000" min_threshold="0.10"/>
+          <FRAG2 frag_size="60000"/>
+          <!-- pbcast.STREAMING_STATE_TRANSFER use_reading_thread="true"/ -->
+          <pbcast.STATE_TRANSFER/>
+          <pbcast.FLUSH timeout="0"/>
+        </config>
+    </stack>
+
+</protocol_stacks>
+
+
Index: src/org/jgroups/util/TCLAction.java
===================================================================
RCS file: src/org/jgroups/util/TCLAction.java
diff -N src/org/jgroups/util/TCLAction.java
--- /dev/null	1 Jan 1970 00:00:00 -0000
+++ src/org/jgroups/util/TCLAction.java	1 Jan 1970 00:00:00 -0000
@@ -0,0 +1,83 @@
+// $Id: Command.java,v 1.3 2006/04/05 05:33:57 belaban Exp $
+
+package org.jgroups.util;
+
+/**
+ * Utility for setting a thread context classloader in the presence
+ * or absence of a SecurityManager. Intended to also work in a J2ME CLDC 
+ * environment where SecurityManager and ClassLoader don't even exist.
+ * 
+ * @author Brian Stansberry
+ *
+ */
+public interface TCLAction {   
+   
+   /**
+    * Sets the thread context classloader for the current thread, or does
+    * nothing if the concept of a TCCL doesn't exist (J2ME).  The classloader
+    * that is set is established when an instance of this interface is
+    * created.
+    */   
+   void setContextClassLoader();
+
+   /**
+    * Sets the thread context classloader for the given thread, or does
+    * nothing if the concept of a TCCL doesn't exist (J2ME).  The classloader
+    * that is set is established when an instance of this interface is
+    * created.
+    * 
+    * @param thread the thread
+    */   
+   void setContextClassLoader(Thread thread);
+   
+   /**
+    * Factory that provides a TCLAction suitable for the runtime environment.
+    */
+   class FACTORY {
+      
+      private static Object getSecurityManagerMethod = null;
+      static {
+         try {
+            getSecurityManagerMethod = java.lang.System.class.getDeclaredMethod("getSecurityManager", new Class[]{});
+         }
+         catch (Throwable e) {
+            // assume this is a CLDC J2ME environment
+            getSecurityManagerMethod = null;
+         }
+      }
+      
+      /**
+       * Returns a TCLAction appropriate for the current security environment,
+       * configured to use the current thread context classloader,
+       * or returns null if setting a thread context classloader is not supported
+       * by the runtime (e.g. J2ME CLDC).
+       * 
+       * @return
+       */
+      public static TCLAction getTCLAction() {
+         if (getSecurityManagerMethod == null)
+             return null;
+         else {            
+             TCLAction action = NonPrivilegedTCLAction.getTCLAction();
+             return action == null ? action : PrivilegedTCLAction.getTCLAction();
+         }         
+      }
+      
+      /**
+       * Returns a TCLAction appropriate for the current security environment,
+       * configured to use the given class' classloader, or returns null
+       * if setting a thread context classloader is not supported
+       * by the runtime (e.g. J2ME CLDC).
+       * 
+       * @return
+       */
+      public static TCLAction getTCLAction(Class source) {
+         if (getSecurityManagerMethod == null)
+             return null;
+         else {            
+             TCLAction action = NonPrivilegedTCLAction.getTCLAction(source);
+             return action == null ? action : PrivilegedTCLAction.getTCLAction(source);
+         }         
+      }
+   }
+}
Index: src/org/jgroups/util/PrivilegedTCLAction.java
===================================================================
RCS file: src/org/jgroups/util/PrivilegedTCLAction.java
diff -N src/org/jgroups/util/PrivilegedTCLAction.java
--- /dev/null	1 Jan 1970 00:00:00 -0000
+++ src/org/jgroups/util/PrivilegedTCLAction.java	1 Jan 1970 00:00:00 -0000
@@ -0,0 +1,70 @@
+package org.jgroups.util;
+
+import java.lang.ref.WeakReference;
+
+/**
+ * TCLAction impl appropriate for when a SecurityManager is in place.
+ * 
+ * @author Brian Stansberry
+ */
+class PrivilegedTCLAction implements TCLAction
+{
+   private final java.lang.ref.WeakReference<ClassLoader> classLoaderRef;
+   
+   /**
+    * Gets a TCLAction that uses the current thread's context classloader as 
+    * the classloader. Will not return <code>null</code>.
+    */
+   static TCLAction getTCLAction()
+   {
+      ClassLoader cl = (ClassLoader) java.security.AccessController.doPrivileged(
+         new java.security.PrivilegedAction()
+         {
+            public Object run()
+            {
+               return Thread.currentThread().getContextClassLoader();
+            }
+         }
+      );
+      
+      return new PrivilegedTCLAction(cl);
+   }
+   
+   /**
+    * Gets a TCLAction that uses the given class' classloader as 
+    * the classloader. Will not return <code>null</code>.
+    */
+   static TCLAction getTCLAction(Class source)
+   {
+      return new PrivilegedTCLAction(source.getClassLoader());
+   }
+   
+   private PrivilegedTCLAction()
+   {
+      this(Thread.currentThread().getContextClassLoader());
+   }
+   
+   private PrivilegedTCLAction(ClassLoader cl) {
+      this.classLoaderRef = new WeakReference(cl);
+   }
+
+   public void setContextClassLoader() {      
+      setContextClassLoader(Thread.currentThread());
+   }
+
+   public void setContextClassLoader(final Thread thread) {
+      ClassLoader ref = classLoaderRef.get();
+      final ClassLoader cl = (ref == null ? getClass().getClassLoader() : ref); 
+      java.security.AccessController.doPrivileged(
+         new java.security.PrivilegedAction()
+         {
+            public Object run()
+            {
+               thread.setContextClassLoader(cl);
+               return null;
+            }
+         }
+      );
+   }
+
+}
Index: src/org/jgroups/util/TCLActionThreadPoolExecutor.java
===================================================================
RCS file: src/org/jgroups/util/TCLActionThreadPoolExecutor.java
diff -N src/org/jgroups/util/TCLActionThreadPoolExecutor.java
--- /dev/null	1 Jan 1970 00:00:00 -0000
+++ src/org/jgroups/util/TCLActionThreadPoolExecutor.java	1 Jan 1970 00:00:00 -0000
@@ -0,0 +1,116 @@
+/*
+ * JBoss, Home of Professional Open Source.
+ * Copyright 2008, Red Hat Middleware LLC, and individual contributors
+ * as indicated by the @author tags. See the copyright.txt file in the
+ * distribution for a full listing of individual contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+
+package org.jgroups.util;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.RejectedExecutionHandler;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * ThreadPoolExecutor subclass that resets the pool thread's context classloader
+ * in the {@link #afterExecute(Runnable, Throwable)} method. Uses a
+ * {@link TCLAction} established in the constructor to do this.
+ * 
+ * @author Brian Stansberry
+ */
+public class TCLActionThreadPoolExecutor extends ThreadPoolExecutor
+{
+   private TCLAction tclAction;
+   
+   /**
+    * {@inheritDoc}
+    * 
+    * @param useTCCL <code>true</code> if the {@link TCLAction} used by this
+    *                pool should use the ClassLoader currently assigned as the
+    *                current thread's context classloader; <code>false</code>
+    *                if it should use this class' classloader.
+    */
+   public TCLActionThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
+         BlockingQueue<Runnable> workQueue, boolean useTCCL)
+   {
+      super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
+      initTCLAction(useTCCL);
+   }
+
+   /**
+    * {@inheritDoc}
+    * 
+    * @param useTCCL <code>true</code> if the {@link TCLAction} used by this
+    *                pool should use the ClassLoader currently assigned as the
+    *                current thread's context classloader; <code>false</code>
+    *                if it should use this class' classloader.
+    */
+   public TCLActionThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
+         BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, boolean useTCCL)
+   {
+      super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
+      initTCLAction(useTCCL);
+   }
+
+   /**
+    * {@inheritDoc}
+    * 
+    * @param useTCCL <code>true</code> if the {@link TCLAction} used by this
+    *                pool should use the ClassLoader currently assigned as the
+    *                current thread's context classloader; <code>false</code>
+    *                if it should use this class' classloader.
+    */
+   public TCLActionThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
+         BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler, boolean useTCCL)
+   {
+      super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
+      initTCLAction(useTCCL);
+   }
+
+   /**
+    * {@inheritDoc}
+    * 
+    * @param useTCCL <code>true</code> if the {@link TCLAction} used by this
+    *                pool should use the ClassLoader currently assigned as the
+    *                current thread's context classloader; <code>false</code>
+    *                if it should use this class' classloader.
+    */
+   public TCLActionThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
+         BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler, boolean useTCCL)
+   {
+      super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
+      initTCLAction(useTCCL);
+   }
+
+   private void initTCLAction(boolean useTCCL)
+   {      
+      this.tclAction = useTCCL ? TCLAction.FACTORY.getTCLAction() : 
+                                 TCLAction.FACTORY.getTCLAction(TCLActionThreadPoolExecutor.class);      
+   }
+
+   protected void afterExecute(Runnable r, Throwable t)
+   {
+      super.afterExecute(r, t);
+      
+      if (tclAction != null)
+         tclAction.setContextClassLoader();
+   }
+
+}
Index: src/org/jgroups/util/NonPrivilegedTCLAction.java
===================================================================
RCS file: src/org/jgroups/util/NonPrivilegedTCLAction.java
diff -N src/org/jgroups/util/NonPrivilegedTCLAction.java
--- /dev/null	1 Jan 1970 00:00:00 -0000
+++ src/org/jgroups/util/NonPrivilegedTCLAction.java	1 Jan 1970 00:00:00 -0000
@@ -0,0 +1,77 @@
+/*
+ * JBoss, Home of Professional Open Source.
+ * Copyright 2008, Red Hat Middleware LLC, and individual contributors
+ * as indicated by the @author tags. See the copyright.txt file in the
+ * distribution for a full listing of individual contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+
+package org.jgroups.util;
+
+import java.lang.ref.WeakReference;
+
+/**
+ * {@link TCLAction} impl appropriate for when no SecurityManager is in place.
+ * 
+ * @author Brian Stansberry
+ */
+class NonPrivilegedTCLAction implements TCLAction
+{
+   private final java.lang.ref.WeakReference<ClassLoader> classLoaderRef;
+   
+   /**
+    * Gets a TCLAction that uses the current thread's context classloader as 
+    * the classloader, or returns null if a SecurityManager is in effect.
+    */
+   static TCLAction getTCLAction()
+   {
+      if (java.lang.System.getSecurityManager() != null)
+         return null;
+      return new NonPrivilegedTCLAction();
+   }
+   
+   /**
+    * Gets a TCLAction that uses the given class' classloader as 
+    * the classloader, or returns null if a SecurityManager is in effect.
+    */
+   static TCLAction getTCLAction(Class source)
+   {
+      if (java.lang.System.getSecurityManager() != null)
+         return null;
+      return new NonPrivilegedTCLAction(source.getClassLoader());
+   }
+   
+   private NonPrivilegedTCLAction()
+   {
+      this(Thread.currentThread().getContextClassLoader());
+   }
+   
+   private NonPrivilegedTCLAction(ClassLoader cl) {
+      this.classLoaderRef = new WeakReference(cl);
+   }
+
+   public void setContextClassLoader() {      
+      setContextClassLoader(Thread.currentThread());
+   }
+
+   public void setContextClassLoader(final Thread thread) {
+      ClassLoader ref = classLoaderRef.get();
+      final ClassLoader cl = (ref == null ? getClass().getClassLoader() : ref); 
+      thread.setContextClassLoader(cl);
+   }
+
+}
Index: tests/junit/org/jgroups/tests/ClassLoaderLeakTest.java
===================================================================
RCS file: tests/junit/org/jgroups/tests/ClassLoaderLeakTest.java
diff -N tests/junit/org/jgroups/tests/ClassLoaderLeakTest.java
--- /dev/null	1 Jan 1970 00:00:00 -0000
+++ tests/junit/org/jgroups/tests/ClassLoaderLeakTest.java	1 Jan 1970 00:00:00 -0000
@@ -0,0 +1,387 @@
+package org.jgroups.tests;
+
+import java.lang.reflect.Method;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+
+import org.jgroups.Address;
+import org.jgroups.Channel;
+import org.jgroups.JChannelFactory;
+import org.jgroups.MembershipListener;
+import org.jgroups.Message;
+import org.jgroups.MessageListener;
+import org.jgroups.View;
+import org.jgroups.blocks.GroupRequest;
+import org.jgroups.blocks.MessageDispatcher;
+import org.jgroups.blocks.RequestHandler;
+
+/**
+ * Tests classloader leak handling of channels.
+ * 
+ * @author Brian Stansberry
+ */
+public class ClassLoaderLeakTest extends ChannelTestBase
+{
+   private static Method STRING_ARG = null;
+   private static Method SIMPLE_MUX = null;
+   private static NoSuchMethodException nsme = null;
+   
+   static
+   {
+      Class clazz = JChannelFactory.class;
+      try
+      {
+         STRING_ARG = clazz.getDeclaredMethod("createChannel", new Class[] { String.class });
+         SIMPLE_MUX = clazz.getDeclaredMethod("createMultiplexerChannel", new Class[] { String.class, String.class });
+      }
+      catch (NoSuchMethodException e)
+      {
+         // Save it for later
+         nsme = e;
+      }
+   }
+   
+   private JChannelFactory factory1;
+   private JChannelFactory factory2;
+   private Channel channel1;
+   private Channel channel2;
+
+   protected void setUp() throws Exception
+   {
+      if (SIMPLE_MUX == null)
+         throw new IllegalStateException("Reflection failed in class init", nsme);
+      
+      super.setUp();
+      
+      factory1 = new JChannelFactory();
+      factory1.setMultiplexerConfig("classloader-leak-test.xml");
+      factory1.setExposeChannels(false);
+      factory1.create();
+      factory1.start();
+      factory2 = new JChannelFactory();
+      factory2.setMultiplexerConfig("classloader-leak-test.xml");
+      factory2.setExposeChannels(false);
+      factory2.create();
+      factory2.start();
+   }
+
+   protected void tearDown() throws Exception
+   {
+      super.tearDown();
+      
+      if (channel1 != null && channel1.isOpen())
+         channel1.close();
+      
+      if (channel2 != null && channel2.isOpen())
+         channel2.close();
+      
+      if (factory1 != null)
+      {
+         factory1.stop();
+         factory1.destroy();
+      }
+      if (factory2 != null)
+      {
+         factory2.stop();
+         factory2.destroy();
+      }
+   }
+   
+   public void testClassLoaderLeakStringShared() throws Exception
+   {  
+      Object[] args1 = { "shared1" };
+      Object[] args2 = { "shared2" };
+      classloaderLeakTest(STRING_ARG, args1, args2);
+   }
+   
+   public void testClassLoaderLeakStringUnshared() throws Exception
+   {  
+      Object[] args = { "unshared" };
+      classloaderLeakTest(STRING_ARG, args, args);
+   }
+   
+   public void testClassLoaderLeakSimpleMuxShared() throws Exception
+   {  
+      Object[] args1 = { "shared1", "leaktest" };
+      Object[] args2 = { "shared2", "leaktest" };
+      classloaderLeakTest(SIMPLE_MUX, args1, args2);
+   }
+   
+   public void testClassLoaderLeakSimpleMuxUnshared() throws Exception
+   {  
+      Object[] args = { "unshared", "leaktest" };
+      classloaderLeakTest(SIMPLE_MUX, args, args);
+   }
+   
+   public void testClassLoaderLeakNonConcurrent() throws Exception
+   {  
+      Object[] args1 = { "nonconcurrent1" };
+      Object[] args2 = { "nonconcurrent2" };
+      classloaderLeakTest(STRING_ARG, args1, args2);
+   }
+   
+   public void testClassLoaderLeakTCP() throws Exception
+   {  
+      Object[] args = { "tcp" };
+      classloaderLeakTest(STRING_ARG, args, args);
+   }
+   
+   private void classloaderLeakTest(Method factoryMeth, Object[] factory1Args, 
+                                    Object[] factory2Args) 
+      throws Exception
+   {
+      // We will create numThreads threads each of which will send
+      // numLoops messages
+      int numThreads = 8;
+      int numLoops = 100;
+      
+      Semaphore semaphore = new Semaphore(numThreads);
+      
+      // We create a ThreadGroup for our test threads so we can identify
+      // test threads that loop back up to the callback methods
+      ThreadGroup runnerGroup = new ThreadGroup("TestRunners");
+      
+      // The classloader we expect our channel threads to use
+      ClassLoader cl = Thread.currentThread().getContextClassLoader();
+      
+      // Handler that receives callbacks from channel, checks that TCCL
+      // is as expected
+      ClassLoaderLeakHandler handler = new ClassLoaderLeakHandler(cl, semaphore, runnerGroup);
+      
+      MessageDispatcher[] dispatchers = new MessageDispatcher[2];
+      
+      // Create our channels, confirm the channel creation callbacks have correct TCCL
+      channel1 = (Channel) factoryMeth.invoke(factory1, factory1Args);
+      dispatchers[0] = new MessageDispatcher(channel1, handler, handler, handler);
+      
+      channel1.connect("leaktest");
+      assertEquals("No classloader leak on channel1 connect", null, handler.getLeakedClassLoader());
+      
+      channel2 = (Channel) factoryMeth.invoke(factory2, factory2Args);
+      dispatchers[1] = new MessageDispatcher(channel2, handler, handler, handler);
+      
+      channel2.connect("leaktest");
+      assertEquals("No classloader leak on channel2 connect", null, handler.getLeakedClassLoader());
+      
+      log.info("Channels connected");
+      
+      // Create runnables that will send messages
+      ClassLoaderLeakRunner[] runners = new ClassLoaderLeakRunner[numThreads];            
+      for (int i = 0; i < runners.length; i++)
+      {
+         // 50% of threads use each channel
+         MessageDispatcher disp = dispatchers[i % 2];            
+         runners[i] = new ClassLoaderLeakRunner(disp, numLoops, runnerGroup, semaphore);         
+      }
+
+      // Make the runnables wait until all are ready
+      semaphore.acquire(numThreads);
+      
+      for (int i = 0; i < runners.length; i++)
+      {
+         runners[i].start();
+      }
+      
+      // Let the runnables go
+      semaphore.release(numThreads);
+      
+      try
+      {
+         assertTrue("messages received within 15 seconds", semaphore.tryAcquire(numThreads, 15, TimeUnit.SECONDS));
+         
+         log.info("Messages received");
+      }
+      finally
+      {
+         for (int i = 0; i < runners.length; i++)
+         {
+            runners[i].stop();
+         }
+      }
+      
+      log.info("Sender threads stopped");
+      
+      assertEquals("No classloader leak", null, handler.getLeakedClassLoader());
+   }
+
+   /**
+    * Runnable that sets the TCCL to a new classloader and loops sending messages.
+    */
+   private class ClassLoaderLeakRunner implements Runnable
+   {
+      private Thread thread;
+      private final MessageDispatcher dispatcher;
+      private final int numMsgs;
+      private final ThreadGroup threadGroup;
+      private final Semaphore semaphore;
+      private boolean stopped;
+      private Exception exception;
+      
+      ClassLoaderLeakRunner(MessageDispatcher dispatcher, int numMsgs, ThreadGroup group, Semaphore semaphore)
+      {
+         this.dispatcher = dispatcher;
+         this.numMsgs = numMsgs;
+         this.threadGroup = group;
+         this.semaphore = semaphore;
+      }
+      
+      public void run()
+      {
+         boolean acquired = false;
+         
+         // Create a new classloader to see if it pollutes a channel thread
+         ClassLoader cl = new ClassLoader(Thread.currentThread().getContextClassLoader()){};
+         try
+         {
+            // block until the main test thread releases us
+            semaphore.acquire();         
+            acquired = true;
+            
+            log.info(Thread.currentThread().getName() + " starting");        
+            
+            Thread.currentThread().setContextClassLoader(cl);
+            
+            for (int i = 0; i < numMsgs && !stopped && !Thread.interrupted(); i++)
+            {
+               Message msg = new Message(null, null, String.valueOf(i));
+               // sending this way calls receive()
+               dispatcher.send(msg);
+               // sending this way calls handle()
+               dispatcher.castMessage(null, msg, GroupRequest.GET_ALL, 0, false);
+            }
+            
+            log.info(Thread.currentThread().getName() + " done");
+         }
+         catch (Exception e)
+         {
+            this.exception = e;
+         }
+         finally
+         {
+            // Signal the main thread that we are done
+            if (acquired)
+               semaphore.release();
+            
+            Thread.currentThread().setContextClassLoader(cl.getParent());
+         }
+      }
+
+      public Exception getException()
+      {
+         return exception;
+      }
+
+      public void start()
+      {
+         thread = new Thread(this.threadGroup, this);
+         thread.setDaemon(true);
+         thread.start();
+      }
+      
+      public void stop()
+      {
+         stopped = true;
+         if (thread != null && thread.isAlive())
+         {
+            try
+            {
+               thread.join(100);
+            }
+            catch (InterruptedException e)
+            {
+            }
+            if (thread.isAlive())
+               thread.interrupt();
+         }
+      }
+   }
+   
+   /**
+    * Implements all the JGroups callback interfaces; checks for the
+    * correct TCCL in all callbacks.
+    */
+   private class ClassLoaderLeakHandler 
+      implements MembershipListener, MessageListener, RequestHandler
+   {
+      private final Semaphore semaphore;
+      private final ClassLoader expected;
+      private final ThreadGroup runnerGroup;
+      private ClassLoader leakedClassLoader;
+      private final int numPermits;
+      
+      ClassLoaderLeakHandler(ClassLoader expected, Semaphore semaphore, ThreadGroup runnerGroup)
+      {
+         this.expected = expected;
+         this.semaphore = semaphore;
+         this.runnerGroup = runnerGroup;
+         this.numPermits = this.semaphore.availablePermits();
+      }
+
+      public Object handle(Message msg)
+      {
+         log.debug("handled(): " + msg.getObject());
+         checkClassLoader(true, "handle()");
+         return null;
+      }
+
+      public void block()
+      {
+         checkClassLoader(false, "block()");
+      }
+
+      public void suspect(Address suspected_mbr)
+      {
+         checkClassLoader(false, "suspect()");         
+      }
+
+      public void viewAccepted(View new_view)
+      {
+         checkClassLoader(false, "viewAccepted()");     
+         log.info("viewAccepted(): " + new_view);
+      }
+
+      public byte[] getState()
+      {
+         checkClassLoader(false, "getState()");  
+         return new byte[1];
+      }
+
+      public void receive(Message msg)
+      {
+         checkClassLoader(false, "receive()");  
+      }
+
+      public void setState(byte[] state)
+      {
+         checkClassLoader(false, "setState()");   
+      }
+
+      public ClassLoader getLeakedClassLoader()
+      {
+         return leakedClassLoader;
+      } 
+      
+      private void checkClassLoader(boolean fromHandle, String method)
+      {         
+         if (leakedClassLoader == null) // ignore msgs once we found a leak
+         {
+            // ignore runner threads that loop all the way back up 
+            if (Thread.currentThread().getThreadGroup().equals(runnerGroup))
+            {               
+               return; 
+            }
+            
+            ClassLoader tccl = Thread.currentThread().getContextClassLoader();
+            if (!expected.equals(tccl))               
+            {
+               leakedClassLoader = tccl;
+               log.info("ClassLoader leaked in " + method + ": " + tccl + " leaked to " + Thread.currentThread().getName());
+               // Release all the permits so the main test thread immediately
+               // wakes up and sees the failure
+               semaphore.release(numPermits);
+            }
+         }         
+      }
+      
+   }
+}
