AMQ-8039 - support system property configuration of the inactivity monitor thread pool, follow example of nio. new test to verify

activemq-client/src/main/java/org/apache/activemq/transport/AbstractInactivityMonitor.java

@@ -18,7 +18,10 @@ package org.apache.activemq.transport;
 
 import java.io.IOException;
 import java.util.Timer;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.RejectedExecutionHandler;
 import java.util.concurrent.SynchronousQueue;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.ThreadPoolExecutor;
@@ -217,8 +220,7 @@ public abstract class AbstractInactivityMonitor extends TransportFilter {
                 });
             } catch (RejectedExecutionException ex) {
                 if (!ASYNC_TASKS.isShutdown()) {
-                    LOG.error("Async write check was rejected from the executor: ", ex);
+                    LOG.warn("Async write check was rejected from the executor: ", ex);
-                    throw ex;
                 }
             }
         } else {
@@ -253,8 +255,7 @@ public abstract class AbstractInactivityMonitor extends TransportFilter {
                 });
             } catch (RejectedExecutionException ex) {
                 if (!ASYNC_TASKS.isShutdown()) {
-                    LOG.error("Async read check was rejected from the executor: ", ex);
+                    LOG.warn("Async read check was rejected from the executor: ", ex);
-                    throw ex;
                 }
             }
         } else {
@@ -501,7 +502,7 @@ public abstract class AbstractInactivityMonitor extends TransportFilter {
                     WRITE_CHECK_TIMER = null;
                     READ_CHECK_TIMER = null;
                     try {
-                        ThreadPoolUtils.shutdownGraceful(ASYNC_TASKS, TimeUnit.SECONDS.toMillis(10));
+                        ThreadPoolUtils.shutdownGraceful(ASYNC_TASKS, 0);
                     } finally {
                         ASYNC_TASKS = null;
                     }
@@ -511,16 +512,18 @@ public abstract class AbstractInactivityMonitor extends TransportFilter {
     }
 
     private final ThreadFactory factory = new ThreadFactory() {
+        private long i = 0;
         @Override
         public Thread newThread(Runnable runnable) {
-            Thread thread = new Thread(runnable, "ActiveMQ InactivityMonitor Worker");
+            Thread thread = new Thread(runnable, "ActiveMQ InactivityMonitor Worker " + (i++));
             thread.setDaemon(true);
             return thread;
         }
     };
 
     private ThreadPoolExecutor createExecutor() {
-        ThreadPoolExecutor exec = new ThreadPoolExecutor(0, Integer.MAX_VALUE, getDefaultKeepAliveTime(), TimeUnit.SECONDS, new SynchronousQueue<Runnable>(), factory);
+        ThreadPoolExecutor exec = new ThreadPoolExecutor(getDefaultCorePoolSize(), getDefaultMaximumPoolSize(), getDefaultKeepAliveTime(),
+                TimeUnit.SECONDS, newWorkQueue(), factory, newRejectionHandler());
         exec.allowCoreThreadTimeOut(true);
         return exec;
     }
@@ -528,4 +531,29 @@ public abstract class AbstractInactivityMonitor extends TransportFilter {
     private static int getDefaultKeepAliveTime() {
         return Integer.getInteger("org.apache.activemq.transport.AbstractInactivityMonitor.keepAliveTime", 30);
     }
+
+    private static int getDefaultCorePoolSize() {
+        return Integer.getInteger("org.apache.activemq.transport.AbstractInactivityMonitor.corePoolSize", 0);
+    }
+
+    private static int getDefaultMaximumPoolSize() {
+        return Integer.getInteger("org.apache.activemq.transport.AbstractInactivityMonitor.maximumPoolSize", Integer.MAX_VALUE);
+    }
+
+    private static int getDefaultWorkQueueCapacity() {
+        return Integer.getInteger("org.apache.activemq.transport.AbstractInactivityMonitor.workQueueCapacity", 0);
+    }
+
+    private static boolean canRejectWork() {
+        return Boolean.getBoolean("org.apache.activemq.transport.AbstractInactivityMonitor.rejectWork");
+    }
+
+    private BlockingQueue<Runnable> newWorkQueue() {
+        final int workQueueCapacity = getDefaultWorkQueueCapacity();
+        return workQueueCapacity > 0 ? new LinkedBlockingQueue<Runnable>(workQueueCapacity) : new SynchronousQueue<Runnable>();
+    }
+
+    private RejectedExecutionHandler newRejectionHandler() {
+        return canRejectWork() ? new ThreadPoolExecutor.AbortPolicy() : new ThreadPoolExecutor.CallerRunsPolicy();
+    }
 }

activemq-unit-tests/src/test/java/org/apache/activemq/transport/RestrictedThreadPoolInactivityTimeoutTest.java

@@ -0,0 +1,141 @@
+/**
+ * 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.activemq.transport;
+
+import junit.framework.Test;
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.JmsTestSupport;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.jmx.ManagementContext;
+import org.apache.activemq.util.SocketProxy;
+import org.apache.activemq.util.URISupport;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.jms.Connection;
+import javax.management.JMException;
+import javax.management.ObjectName;
+import java.net.URI;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+public class RestrictedThreadPoolInactivityTimeoutTest extends JmsTestSupport {
+    private static final Logger LOG = LoggerFactory.getLogger(RestrictedThreadPoolInactivityTimeoutTest.class);
+
+    public String brokerTransportScheme = "tcp";
+    public Boolean rejectWork = Boolean.FALSE;
+
+    final int poolSize = 2;
+    final int numConnections = 10;
+    final CountDownLatch doneOneConnectionAddress = new CountDownLatch(1);
+    final CountDownLatch doneConsumers = new CountDownLatch(numConnections);
+
+    protected BrokerService createBroker() throws Exception {
+
+        if (rejectWork) {
+            System.setProperty("org.apache.activemq.transport.AbstractInactivityMonitor.workQueueCapacity", Integer.toString(poolSize));
+            System.setProperty("org.apache.activemq.transport.AbstractInactivityMonitor.rejectWork", "true");
+        }
+        System.setProperty("org.apache.activemq.transport.AbstractInactivityMonitor.maximumPoolSize", Integer.toString(poolSize));
+
+        BrokerService broker = super.createBroker();
+        broker.setPersistent(false);
+        broker.setUseJmx(true);
+        broker.setManagementContext(new ManagementContext() {
+            @Override
+            public void unregisterMBean(ObjectName name) throws JMException {
+                if (name.getKeyPropertyListString().contains("remoteAddress")) {
+                    // a client connection mbean, removed by inactivity monitor task
+                    // simulate a slow mbean unregister
+                    LOG.info("SLEEP : " + Thread.currentThread() + ": on remoteAddress unregister: " + name);
+                    try {
+                        TimeUnit.SECONDS.sleep(2);
+                    } catch (InterruptedException ok) {
+                    }
+                    doneOneConnectionAddress.countDown();
+                } else if (name.getKeyPropertyListString().contains("Consumer")) {
+                    // consumer removal from asyncStop task, this is blocked on service lock
+                    // during inactivity monitor onException
+                    LOG.info(Thread.currentThread() + ": on consumer unregister: " + name);
+                    doneConsumers.countDown();
+                }
+                super.unregisterMBean(name);
+            }
+        });
+
+        broker.addConnector(brokerTransportScheme + "://localhost:0?wireFormat.maxInactivityDuration=1000&wireFormat.maxInactivityDurationInitalDelay=1000");
+        return broker;
+    }
+
+    public void initCombosForTestThreadsInvolvedInXInactivityTimeouts() {
+        addCombinationValues("brokerTransportScheme", new Object[]{"tcp", "nio"});
+        addCombinationValues("rejectWork", new Object[] {Boolean.TRUE, Boolean.FALSE});
+    }
+
+    public void testThreadsInvolvedInXInactivityTimeouts() throws Exception {
+
+        URI tcpBrokerUri = URISupport.removeQuery(broker.getTransportConnectors().get(0).getConnectUri());
+
+        SocketProxy proxy = new SocketProxy();
+        proxy.setTarget(tcpBrokerUri);
+        proxy.open();
+
+        // leave the server to do the only inactivity monitoring
+        URI clientUri =  URISupport.createURIWithQuery(proxy.getUrl(), "useInactivityMonitor=false");
+        LOG.info("using server uri: " + tcpBrokerUri + ", client uri: " + clientUri);
+
+        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(clientUri);
+
+        for (int i=0; i<numConnections;i++) {
+            Connection c = factory.createConnection();
+            c.start();
+        }
+
+        proxy.pause();
+
+        int before = Thread.currentThread().getThreadGroup().activeCount();
+        LOG.info("threads before: " + before);
+
+        // expect inactivity monitor to kick in after 2*timeout
+
+        Thread.yield();
+
+        // after one sleep, unbounded pools will have filled with threads
+        doneOneConnectionAddress.await(10, TimeUnit.SECONDS);
+
+        int after = Thread.currentThread().getThreadGroup().activeCount();
+
+        int diff = Math.abs(before - after);
+        LOG.info("threads after: " + after + ", diff: " + diff);
+
+        assertTrue("Should be at most inactivity monitor pool size * 2. Diff = " + diff, diff <= 2*poolSize);
+
+        assertTrue("all work complete", doneConsumers.await(10, TimeUnit.SECONDS));
+    }
+
+    @Override
+    protected void tearDown() throws Exception {
+        super.tearDown();
+        System.clearProperty("org.apache.activemq.transport.AbstractInactivityMonitor.workQueueCapacity");
+        System.clearProperty("org.apache.activemq.transport.AbstractInactivityMonitor.maximumPoolSize");
+        System.clearProperty("org.apache.activemq.transport.AbstractInactivityMonitor.rejectWork");
+    }
+
+    public static Test suite() {
+        return suite(RestrictedThreadPoolInactivityTimeoutTest.class);
+    }
+}

activemq-unit-tests/src/test/java/org/apache/activemq/transport/tcp/InactivityMonitorTest.java

@@ -24,6 +24,7 @@ import java.util.concurrent.atomic.AtomicInteger;
 
 import javax.net.SocketFactory;
 
+import junit.framework.Test;
 import org.apache.activemq.CombinationTestSupport;
 import org.apache.activemq.command.WireFormatInfo;
 import org.apache.activemq.openwire.OpenWireFormat;
@@ -55,6 +56,11 @@ public class InactivityMonitorTest extends CombinationTestSupport implements Tra
     private final AtomicBoolean ignoreClientError = new AtomicBoolean(false);
     private final AtomicBoolean ignoreServerError = new AtomicBoolean(false);
 
+
+    public static Test suite() {
+         return suite(InactivityMonitorTest.class);
+    }
+
     @Override
     protected void setUp() throws Exception {
         super.setUp();
@@ -146,7 +152,7 @@ public class InactivityMonitorTest extends CombinationTestSupport implements Tra
 
                 @Override
                 public void onException(IOException error) {
-                    if (!ignoreClientError.get()) {
+                    if (!ignoreServerError.get()) {
                         LOG.info("Server transport error:", error);
                         serverErrorCount.incrementAndGet();
                     }
@@ -239,7 +245,6 @@ public class InactivityMonitorTest extends CombinationTestSupport implements Tra
      * @throws URISyntaxException
      */
     public void initCombosForTestNoClientHangWithServerBlock() throws Exception {
-        startClient();
 
         addCombinationValues("clientInactivityLimit", new Object[] {Long.valueOf(1000)});
         addCombinationValues("serverInactivityLimit", new Object[] {Long.valueOf(1000)});