ARTEMIS-4723 Avoid objects left on ThreadLocal from OrderedExecutorFactory

co-authored: Jakob van Kruijssen <cardamon@gmail.com>

artemis-commons/src/main/java/org/apache/activemq/artemis/utils/actors/HandlerBase.java

@@ -26,22 +26,39 @@ package org.apache.activemq.artemis.utils.actors;
 public abstract class HandlerBase {
 
    private static class Counter {
-      int count = 0;
+      int count = 1;
    }
 
    /** an actor could be used within an OrderedExecutor. So we need this counter to decide if there's a Handler anywhere in the stack trace */
-   private static final ThreadLocal<Counter> inHandler = ThreadLocal.withInitial(() -> new Counter());
+   private static final ThreadLocal<Counter> counterThreadLocal = new ThreadLocal<>();
 
    protected static void enter() {
-      inHandler.get().count++;
+      Counter counter = counterThreadLocal.get();
+      if (counter == null) {
+         counter = new Counter(); // it starts at 1, so no need to increment it
+         counterThreadLocal.set(counter);
+      } else {
+         counter.count++;
+      }
    }
 
    public static boolean inHandler() {
-      return inHandler.get().count > 0;
+      Counter counter = counterThreadLocal.get();
+      if (counter == null) {
+         return false;
+      } else if (counter.count == 0) {
+         counterThreadLocal.remove();
+      }
+      return counter.count > 0;
    }
 
    protected static void leave() {
-      inHandler.get().count--;
+      Counter counter = counterThreadLocal.get();
+      if (counter != null) {
+         if (--counter.count <= 0) {
+            counterThreadLocal.remove();
+         }
+      }
    }
 
 }

tests/leak-tests/src/test/java/org/apache/activemq/artemis/tests/leak/AbstractLeakTest.java

@@ -0,0 +1,41 @@
+/*
+ * 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.artemis.tests.leak;
+
+import io.github.checkleak.core.CheckLeak;
+import org.apache.activemq.artemis.core.journal.impl.JournalImpl;
+import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
+import org.apache.activemq.artemis.core.server.impl.ServerStatus;
+import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
+import org.apache.activemq.artemis.utils.actors.OrderedExecutor;
+import org.junit.AfterClass;
+
+public abstract class AbstractLeakTest extends ActiveMQTestBase {
+
+   @AfterClass
+   public static void clearStatus() throws Exception {
+      ServerStatus.clear();
+
+      CheckLeak checkLeak = new CheckLeak();
+      MemoryAssertions.assertMemory(checkLeak, 0, JournalImpl.class.getName());
+      MemoryAssertions.assertMemory(checkLeak, 0, ActiveMQServerImpl.class.getName());
+      MemoryAssertions.assertMemory(checkLeak, 0, OrderedExecutor.class.getName());
+      MemoryAssertions.assertNoInnerInstances(OrderedExecutor.class, checkLeak);
+   }
+
+}

tests/leak-tests/src/test/java/org/apache/activemq/artemis/tests/leak/ClientLeakTest.java

@@ -30,7 +30,6 @@ import org.apache.activemq.artemis.api.core.TransportConfiguration;
 import org.apache.activemq.artemis.core.config.impl.ConfigurationImpl;
 import org.apache.activemq.artemis.core.server.ActiveMQServer;
 import org.apache.activemq.artemis.core.server.ActiveMQServers;
-import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
 import org.apache.activemq.artemis.tests.util.CFUtil;
 import org.apache.activemq.artemis.utils.SpawnedVMSupport;
 import org.apache.qpid.proton.engine.impl.ReceiverImpl;
@@ -47,7 +46,7 @@ import static org.apache.activemq.artemis.tests.leak.MemoryAssertions.assertMemo
 
 // This test spawns the server as a separate VM
 // as we need to count exclusively client objects from qpid-proton
-public class ClientLeakTest extends ActiveMQTestBase {
+public class ClientLeakTest extends AbstractLeakTest {
 
    private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
 

tests/leak-tests/src/test/java/org/apache/activemq/artemis/tests/leak/ConnectionDroppedLeakTest.java

@@ -40,7 +40,6 @@ import org.apache.activemq.artemis.core.server.Queue;
 import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
 import org.apache.activemq.artemis.core.server.impl.AddressInfo;
 import org.apache.activemq.artemis.core.server.impl.ServerStatus;
-import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
 import org.apache.activemq.artemis.tests.util.CFUtil;
 import org.junit.After;
 import org.junit.Assert;
@@ -54,7 +53,7 @@ import org.slf4j.LoggerFactory;
 import static org.apache.activemq.artemis.tests.leak.MemoryAssertions.assertMemory;
 import static org.apache.activemq.artemis.tests.leak.MemoryAssertions.basicMemoryAsserts;
 
-public class ConnectionDroppedLeakTest extends ActiveMQTestBase {
+public class ConnectionDroppedLeakTest extends AbstractLeakTest {
 
    private ConnectionFactory createConnectionFactory(String protocol) {
       if (protocol.equals("AMQP")) {

tests/leak-tests/src/test/java/org/apache/activemq/artemis/tests/leak/ConnectionLeakTest.java

@@ -43,7 +43,6 @@ import org.apache.activemq.artemis.core.server.impl.ServerConsumerImpl;
 import org.apache.activemq.artemis.core.server.impl.ServerStatus;
 import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;
 import org.apache.activemq.artemis.protocol.amqp.broker.AMQPStandardMessage;
-import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
 import org.apache.activemq.artemis.tests.util.CFUtil;
 import org.apache.activemq.artemis.utils.Wait;
 import org.apache.activemq.artemis.utils.collections.LinkedListImpl;
@@ -60,7 +59,7 @@ import org.slf4j.LoggerFactory;
 import static org.apache.activemq.artemis.tests.leak.MemoryAssertions.assertMemory;
 import static org.apache.activemq.artemis.tests.leak.MemoryAssertions.basicMemoryAsserts;
 
-public class ConnectionLeakTest extends ActiveMQTestBase {
+public class ConnectionLeakTest extends AbstractLeakTest {
 
    private ConnectionFactory createConnectionFactory(String protocol) {
       if (protocol.equals("AMQP")) {

tests/leak-tests/src/test/java/org/apache/activemq/artemis/tests/leak/CoreClientLeakTest.java

@@ -24,16 +24,21 @@ import org.apache.activemq.artemis.api.core.client.ClientConsumer;
 import org.apache.activemq.artemis.api.core.client.ClientSession;
 import org.apache.activemq.artemis.api.core.client.ClientSessionFactory;
 import org.apache.activemq.artemis.api.core.client.ServerLocator;
+import org.apache.activemq.artemis.core.journal.impl.JournalImpl;
 import org.apache.activemq.artemis.core.server.ActiveMQServer;
-import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
+import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
+import org.apache.activemq.artemis.core.server.impl.ServerStatus;
 import org.apache.activemq.artemis.utils.RandomUtil;
+import org.apache.activemq.artemis.utils.actors.OrderedExecutor;
+import org.junit.After;
+import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.Assume;
 import org.junit.Before;
 import org.junit.BeforeClass;
 import org.junit.Test;
 
-public class CoreClientLeakTest extends ActiveMQTestBase {
+public class CoreClientLeakTest extends AbstractLeakTest {
 
    ActiveMQServer server;
 
@@ -42,6 +47,14 @@ public class CoreClientLeakTest extends ActiveMQTestBase {
       Assume.assumeTrue(CheckLeak.isLoaded());
    }
 
+   @AfterClass
+   public static void afterClass() throws Exception {
+      ServerStatus.clear();
+      MemoryAssertions.assertMemory(new CheckLeak(), 0, ActiveMQServerImpl.class.getName());
+      MemoryAssertions.assertMemory(new CheckLeak(), 0, JournalImpl.class.getName());
+      MemoryAssertions.assertMemory(new CheckLeak(), 0, OrderedExecutor.class.getName());
+   }
+
    @Override
    @Before
    public void setUp() throws Exception {
@@ -51,6 +64,14 @@ public class CoreClientLeakTest extends ActiveMQTestBase {
       server.start();
    }
 
+   @Override
+   @After
+   public void tearDown() throws Exception {
+      super.tearDown();
+      server.stop();
+      server = null;
+   }
+
    @Test
    public void testConsumerFiltered() throws Exception {
 
@@ -95,6 +116,9 @@ public class CoreClientLeakTest extends ActiveMQTestBase {
             clientSession.deleteQueue(queue);
          }
       }
+
+      locator.close();
    }
 
+
 }

tests/leak-tests/src/test/java/org/apache/activemq/artemis/tests/leak/HandlerBaseLeakTest.java

@@ -0,0 +1,59 @@
+/*
+ * 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.artemis.tests.leak;
+
+import java.lang.invoke.MethodHandles;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import io.github.checkleak.core.CheckLeak;
+import org.apache.activemq.artemis.utils.actors.OrderedExecutor;
+import org.apache.activemq.artemis.utils.actors.OrderedExecutorFactory;
+import org.junit.Assert;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class HandlerBaseLeakTest extends AbstractLeakTest {
+
+   private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+   @Test
+   public void testHandlerLeak() throws Throwable {
+      ExecutorService service = Executors.newFixedThreadPool(1);
+      runAfter(service::shutdownNow);
+      executeFactory(service);
+
+      CheckLeak checkLeak = new CheckLeak();
+
+      MemoryAssertions.assertNoInnerInstances(OrderedExecutor.class, checkLeak);
+   }
+
+   // this needs to be a sub-method, to make sure the VM will collect and discard certain objects, Otherwise HandlerBase would still show in the heap
+   private static void executeFactory(ExecutorService service) throws InterruptedException {
+      OrderedExecutorFactory factory = new OrderedExecutorFactory(service);
+      CountDownLatch latch = new CountDownLatch(1);
+      Executor executor = factory.getExecutor();
+      executor.execute(latch::countDown);
+      Assert.assertTrue(latch.await(1, TimeUnit.MINUTES));
+   }
+
+}

tests/leak-tests/src/test/java/org/apache/activemq/artemis/tests/leak/JournalLeakTest.java

@@ -38,7 +38,6 @@ import org.apache.activemq.artemis.core.protocol.core.impl.RemotingConnectionImp
 import org.apache.activemq.artemis.core.server.ActiveMQServer;
 import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
 import org.apache.activemq.artemis.core.server.impl.ServerStatus;
-import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
 import org.apache.activemq.artemis.tests.util.CFUtil;
 import org.junit.After;
 import org.junit.Assert;
@@ -53,7 +52,7 @@ import static org.apache.activemq.artemis.tests.leak.MemoryAssertions.assertMemo
 import static org.apache.activemq.artemis.tests.leak.MemoryAssertions.basicMemoryAsserts;
 
 /* at the time this test was written JournalFileImpl was leaking through JournalFileImpl::negative creating a linked list (or leaked-list, pun intended) */
-public class JournalLeakTest extends ActiveMQTestBase {
+public class JournalLeakTest extends AbstractLeakTest {
 
    private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
 

tests/leak-tests/src/test/java/org/apache/activemq/artemis/tests/leak/LinkedListMemoryTest.java

@@ -28,7 +28,7 @@ import org.junit.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-public class LinkedListMemoryTest {
+public class LinkedListMemoryTest extends AbstractLeakTest {
 
    private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
 

tests/leak-tests/src/test/java/org/apache/activemq/artemis/tests/leak/MemoryAssertions.java

@@ -17,9 +17,15 @@
 
 package org.apache.activemq.artemis.tests.leak;
 
+import java.io.PrintWriter;
+import java.io.StringWriter;
 import java.lang.invoke.MethodHandles;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
 
 import io.github.checkleak.core.CheckLeak;
+import io.github.checkleak.core.InventoryDataPoint;
 import org.apache.activemq.artemis.core.protocol.core.impl.RemotingConnectionImpl;
 import org.apache.activemq.artemis.core.protocol.openwire.OpenWireConnection;
 import org.apache.activemq.artemis.core.server.impl.MessageReferenceImpl;
@@ -81,4 +87,42 @@ public class MemoryAssertions {
       }
    }
 
+
+   public static void assertNoInnerInstances(Class clazz, CheckLeak checkLeak) throws Exception {
+      checkLeak.forceGC();
+      List<String> classList = getClassList(clazz);
+
+      Map<Class<?>, InventoryDataPoint> inventoryDataPointMap = checkLeak.produceInventory();
+
+      StringWriter stringWriter = new StringWriter();
+      PrintWriter printWriter = new PrintWriter(stringWriter);
+      boolean failed = false;
+
+      for (Map.Entry<Class<?>, InventoryDataPoint> entry : inventoryDataPointMap.entrySet()) {
+         for (String classElement : classList) {
+            if (entry.getKey().getName().startsWith(classElement) && entry.getValue().getInstances() > 0) {
+               failed = true;
+               printWriter.println(entry.getKey() + " contains " + entry.getValue().getInstances() + " instances");
+               logger.warn("references: towards {}: {}", entry.getKey().getName(), checkLeak.exploreObjectReferences(entry.getKey().getName(), 10, 20, true));
+            }
+         }
+      }
+
+      Assert.assertFalse(stringWriter.toString(), failed);
+   }
+
+   private static List<String> getClassList(Class clazz) {
+      List<String> classList = new ArrayList<>();
+      classList.add(clazz.getName());
+
+      Class<?> superclass = clazz.getSuperclass();
+      while (superclass != null) {
+         if (superclass != Object.class) {
+            classList.add(superclass.getName());
+         }
+         superclass = superclass.getSuperclass();
+      }
+      return classList;
+   }
+
 }

tests/leak-tests/src/test/java/org/apache/activemq/artemis/tests/leak/MessageReferenceLeakTest.java

@@ -30,7 +30,6 @@ import org.apache.activemq.artemis.api.core.client.ServerLocator;
 import org.apache.activemq.artemis.core.server.ActiveMQServer;
 import org.apache.activemq.artemis.core.server.MessageReference;
 import org.apache.activemq.artemis.core.server.Queue;
-import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
 import org.apache.activemq.artemis.utils.RandomUtil;
 import org.junit.Assume;
 import org.junit.Before;
@@ -41,7 +40,7 @@ import org.slf4j.LoggerFactory;
 
 import java.lang.invoke.MethodHandles;
 
-public class MessageReferenceLeakTest extends ActiveMQTestBase {
+public class MessageReferenceLeakTest extends AbstractLeakTest {
 
    private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
 

tests/leak-tests/src/test/java/org/apache/activemq/artemis/tests/leak/ProducerBlockedLeakTest.java

@@ -38,7 +38,6 @@ import org.apache.activemq.artemis.core.server.impl.AddressInfo;
 import org.apache.activemq.artemis.core.settings.impl.AddressFullMessagePolicy;
 import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
 import org.apache.activemq.artemis.logs.AssertionLoggerHandler;
-import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
 import org.apache.activemq.artemis.tests.util.CFUtil;
 import org.apache.activemq.artemis.utils.SpawnedVMSupport;
 import org.apache.activemq.artemis.utils.Wait;
@@ -50,7 +49,7 @@ import org.junit.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-public class ProducerBlockedLeakTest extends ActiveMQTestBase {
+public class ProducerBlockedLeakTest extends AbstractLeakTest {
 
    private static final int OK = 100;
 

tests/leak-tests/src/test/java/org/apache/activemq/artemis/tests/leak/RedistributorLeakTest.java

@@ -34,7 +34,6 @@ import org.apache.activemq.artemis.core.server.ActiveMQServer;
 import org.apache.activemq.artemis.core.server.impl.AddressInfo;
 import org.apache.activemq.artemis.core.server.impl.MessageReferenceImpl;
 import org.apache.activemq.artemis.core.server.impl.QueueImpl;
-import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
 import org.apache.activemq.artemis.tests.util.CFUtil;
 import org.apache.activemq.artemis.utils.RandomUtil;
 import org.apache.activemq.artemis.utils.collections.LinkedListImpl;
@@ -46,7 +45,7 @@ import org.junit.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-public class RedistributorLeakTest extends ActiveMQTestBase {
+public class RedistributorLeakTest extends AbstractLeakTest {
 
    private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
 

tests/leak-tests/src/test/java/org/apache/activemq/artemis/tests/leak/RefCountMessageLeakTest.java

@@ -22,13 +22,12 @@ import java.util.concurrent.TimeUnit;
 import io.github.checkleak.core.CheckLeak;
 import org.apache.activemq.artemis.api.core.RefCountMessage;
 import org.apache.activemq.artemis.api.core.RefCountMessageAccessor;
-import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
 import org.apache.activemq.artemis.utils.RandomUtil;
 import org.apache.activemq.artemis.utils.ReusableLatch;
 import org.junit.Assert;
 import org.junit.Test;
 
-public class RefCountMessageLeakTest extends ActiveMQTestBase {
+public class RefCountMessageLeakTest extends AbstractLeakTest {
 
    static class DebugMessage extends RefCountMessage {
       final String string;