This is an automated email from the ASF dual-hosted git repository.
clebertsuconic pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git
The following commit(s) were added to refs/heads/main by this push:
new 9e524d978d ARTEMIS-4175 JournalFileImpl Leaking
9e524d978d is described below
commit 9e524d978d78f24e4c036e573fd2814fb48caf63
Author: Clebert Suconic <[email protected]>
AuthorDate: Sat Feb 18 07:49:20 2023 -0500
ARTEMIS-4175 JournalFileImpl Leaking
I am now removing the negatives once the file is removed or reused
---
.../artemis/core/journal/impl/JournalFile.java | 6 +
.../artemis/core/journal/impl/JournalFileImpl.java | 5 +
.../core/journal/impl/JournalFilesRepository.java | 7 +
.../artemis/tests/leak/ConnectionLeakTest.java | 1 +
.../artemis/tests/leak/JournalLeakTest.java | 184 +++++++++++++++++++++
.../artemis/tests/leak/MemoryAssertions.java | 2 +-
.../unit/core/journal/impl/ReclaimerTest.java | 4 +
7 files changed, 208 insertions(+), 1 deletion(-)
diff --git
a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/impl/JournalFile.java
b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/impl/JournalFile.java
index 264962697b..f7048d844b 100644
---
a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/impl/JournalFile.java
+++
b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/impl/JournalFile.java
@@ -58,6 +58,12 @@ public interface JournalFile {
*/
int getTotalNegativeToOthers();
+ /**
+ * Callback for when a file is removed. to cleanup negatives and avoid
leaks.
+ * @param fileRemoved
+ */
+ void fileRemoved(JournalFile fileRemoved);
+
/**
* Whether this file additions all have a delete in some other file
*/
diff --git
a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/impl/JournalFileImpl.java
b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/impl/JournalFileImpl.java
index 56cc85385b..4b6456943d 100644
---
a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/impl/JournalFileImpl.java
+++
b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/impl/JournalFileImpl.java
@@ -249,4 +249,9 @@ public class JournalFileImpl implements JournalFile {
return totalNegativeToOthers.get();
}
+ @Override
+ public void fileRemoved(JournalFile fileRemoved) {
+ negCounts.remove(fileRemoved);
+ }
+
}
diff --git
a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/impl/JournalFilesRepository.java
b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/impl/JournalFilesRepository.java
index 3767e87cab..005a8a36bc 100644
---
a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/impl/JournalFilesRepository.java
+++
b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/impl/JournalFilesRepository.java
@@ -248,6 +248,7 @@ public class JournalFilesRepository {
if (!dataFiles.remove(file)) {
ActiveMQJournalLogger.LOGGER.couldNotRemoveFile(file);
}
+ removeNegatives(file);
}
public int getDataFilesCount() {
@@ -728,9 +729,15 @@ public class JournalFilesRepository {
sf.close(false, false);
+ removeNegatives(file);
+
return jf;
}
+ public void removeNegatives(final JournalFile file) {
+ dataFiles.forEach(f -> f.fileRemoved(file));
+ }
+
@Override
public String toString() {
return "JournalFilesRepository(dataFiles=" + dataFiles + ", freeFiles="
+ freeFiles + ", openedFiles=" +
diff --git
a/tests/leak-tests/src/test/java/org/apache/activemq/artemis/tests/leak/ConnectionLeakTest.java
b/tests/leak-tests/src/test/java/org/apache/activemq/artemis/tests/leak/ConnectionLeakTest.java
index 6aab2271bf..9cbcda36f6 100644
---
a/tests/leak-tests/src/test/java/org/apache/activemq/artemis/tests/leak/ConnectionLeakTest.java
+++
b/tests/leak-tests/src/test/java/org/apache/activemq/artemis/tests/leak/ConnectionLeakTest.java
@@ -82,6 +82,7 @@ public class ConnectionLeakTest extends ActiveMQTestBase {
@Before
public void setUp() throws Exception {
server = createServer(true, createDefaultConfig(1, true));
+ server.getConfiguration().setJournalPoolFiles(4).setJournalMinFiles(2);
server.start();
}
diff --git
a/tests/leak-tests/src/test/java/org/apache/activemq/artemis/tests/leak/JournalLeakTest.java
b/tests/leak-tests/src/test/java/org/apache/activemq/artemis/tests/leak/JournalLeakTest.java
new file mode 100644
index 0000000000..4ca1a633e5
--- /dev/null
+++
b/tests/leak-tests/src/test/java/org/apache/activemq/artemis/tests/leak/JournalLeakTest.java
@@ -0,0 +1,184 @@
+/*
+ * 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 javax.jms.BytesMessage;
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import java.lang.invoke.MethodHandles;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import io.github.checkleak.core.CheckLeak;
+import org.apache.activemq.artemis.core.journal.impl.JournalFileImpl;
+import org.apache.activemq.artemis.core.journal.impl.JournalImpl;
+import
org.apache.activemq.artemis.core.persistence.impl.journal.JournalStorageManager;
+import
org.apache.activemq.artemis.core.protocol.core.impl.RemotingConnectionImpl;
+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;
+import org.junit.Assume;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static
org.apache.activemq.artemis.tests.leak.MemoryAssertions.assertMemory;
+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 {
+
+ private static final Logger logger =
LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ ActiveMQServer server;
+
+ @BeforeClass
+ public static void beforeClass() throws Exception {
+ Assume.assumeTrue(CheckLeak.isLoaded());
+ }
+
+ @After
+ public void validateServer() throws Exception {
+ CheckLeak checkLeak = new CheckLeak();
+
+ // I am doing this check here because the test method might hold a
client connection
+ // so this check has to be done after the test, and before the server is
stopped
+ assertMemory(checkLeak, 0, RemotingConnectionImpl.class.getName());
+
+ server.stop();
+
+ server = null;
+
+ clearServers();
+ ServerStatus.clear();
+
+ assertMemory(checkLeak, 0, ActiveMQServerImpl.class.getName());
+ }
+
+ @Override
+ @Before
+ public void setUp() throws Exception {
+ server = createServer(true, createDefaultConfig(1, true));
+ server.getConfiguration().setJournalPoolFiles(4).setJournalMinFiles(2);
+ server.start();
+ }
+
+ @Test
+ public void testAMQP() throws Exception {
+ doTest("AMQP");
+ }
+
+ @Test
+ public void testCore() throws Exception {
+ doTest("CORE");
+ }
+
+ @Test
+ public void testOpenWire() throws Exception {
+ doTest("OPENWIRE");
+ }
+
+ private void doTest(String protocol) throws Exception {
+ int MESSAGES = 10000;
+ int MESSAGE_SIZE = 104;
+ basicMemoryAsserts();
+
+ ExecutorService executorService = Executors.newFixedThreadPool(2);
+ runAfter(executorService::shutdownNow);
+
+ ConnectionFactory cf = CFUtil.createConnectionFactory(protocol,
"tcp://localhost:61616");
+
+ final AtomicInteger errors = new AtomicInteger(0);
+ CountDownLatch done = new CountDownLatch(2);
+
+ executorService.execute(() -> {
+ try {
+ try (Connection connection = cf.createConnection()) {
+ Session session = connection.createSession(true,
Session.SESSION_TRANSACTED);
+ MessageConsumer consumer =
session.createConsumer(session.createQueue(getName()));
+ connection.start();
+
+ for (int i = 0; i < MESSAGES; i++) {
+ Message message = consumer.receive(5000);
+ Assert.assertNotNull(message);
+ Assert.assertEquals(i, message.getIntProperty("i"));
+ if (i > 0 && i % 100 == 0) {
+ session.commit();
+ }
+ }
+ session.commit();
+ }
+ } catch (Throwable e) {
+ errors.incrementAndGet();
+ logger.warn(e.getMessage(), e);
+ } finally {
+ done.countDown();
+ }
+ });
+
+ executorService.execute(() -> {
+ try {
+ try (Connection connection = cf.createConnection()) {
+ Session session = connection.createSession(true,
Session.SESSION_TRANSACTED);
+ MessageProducer producer =
session.createProducer(session.createQueue(getName()));
+ connection.start();
+
+ for (int i = 0; i < MESSAGES; i++) {
+ BytesMessage message = session.createBytesMessage();
+ message.writeBytes(new byte[MESSAGE_SIZE]);
+ message.setIntProperty("i", i);
+ producer.send(message);
+ if (i > 0 && i % 100 == 0) {
+ session.commit();
+ }
+ }
+ session.commit();
+ }
+ } catch (Throwable e) {
+ errors.incrementAndGet();
+ logger.warn(e.getMessage(), e);
+ } finally {
+ done.countDown();
+ }
+ });
+
+ Assert.assertTrue(done.await(1, TimeUnit.MINUTES));
+ Assert.assertEquals(0, errors.get());
+
+ basicMemoryAsserts();
+
+ JournalStorageManager journalStorageManager = (JournalStorageManager)
server.getStorageManager();
+ JournalImpl journalImpl = (JournalImpl)
journalStorageManager.getMessageJournal();
+ int totalFiles = journalImpl.getFilesRepository().getDataFiles().size()
+ journalImpl.getFilesRepository().getFreeFilesCount() +
journalImpl.getOpenedFilesCount();
+
+ // on this particular leak we would have 100 files. I am allowing a
little cushion as they will be from currentFile and some other async opens
+ assertMemory(new CheckLeak(), totalFiles + 5,
JournalFileImpl.class.getName());
+ }
+}
\ No newline at end of file
diff --git
a/tests/leak-tests/src/test/java/org/apache/activemq/artemis/tests/leak/MemoryAssertions.java
b/tests/leak-tests/src/test/java/org/apache/activemq/artemis/tests/leak/MemoryAssertions.java
index 931488e4b9..3e5eac93ea 100644
---
a/tests/leak-tests/src/test/java/org/apache/activemq/artemis/tests/leak/MemoryAssertions.java
+++
b/tests/leak-tests/src/test/java/org/apache/activemq/artemis/tests/leak/MemoryAssertions.java
@@ -56,7 +56,7 @@ public class MemoryAssertions {
for (Object obj : objects) {
logger.warn("Object {} still in the heap", obj);
}
- String report = checkLeak.exploreObjectReferences(5, 10, true,
objects);
+ String report = checkLeak.exploreObjectReferences(10, 10, true,
objects);
logger.info(report);
Assert.fail("Class " + clazz + " has leaked " + objects.length + "
objects\n" + report);
diff --git
a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/journal/impl/ReclaimerTest.java
b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/journal/impl/ReclaimerTest.java
index a2b09975ab..fdae381786 100644
---
a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/journal/impl/ReclaimerTest.java
+++
b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/journal/impl/ReclaimerTest.java
@@ -744,6 +744,10 @@ public class ReclaimerTest extends ActiveMQTestBase {
}
}
+ @Override
+ public void fileRemoved(JournalFile fileRemoved) {
+ }
+
@Override
public int getReplaceableCount() {
return 0;
