From f0d2e9326d241db8c49d86f85c79d6e3ce0d8b8a Mon Sep 17 00:00:00 2001 From: Arnold Galovics Date: Tue, 28 Mar 2023 11:28:44 +0200 Subject: [PATCH] FINERACT-1724: Closing JMS sessions explicitly to avoid memory leaking for multi external event producer --- .../jms/JMSMultiExternalEventProducer.java | 30 +++++++++++++++---- .../JMSMultiExternalEventProducerTest.java | 8 +++++ 2 files changed, 32 insertions(+), 6 deletions(-) diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/producer/jms/JMSMultiExternalEventProducer.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/producer/jms/JMSMultiExternalEventProducer.java index 3511780d8..15b2edcb2 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/producer/jms/JMSMultiExternalEventProducer.java +++ b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/producer/jms/JMSMultiExternalEventProducer.java @@ -26,6 +26,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.Future; +import java.util.stream.Collectors; import javax.jms.Connection; import javax.jms.ConnectionFactory; import javax.jms.Destination; @@ -34,6 +35,8 @@ import javax.jms.MessageProducer; import javax.jms.Session; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.tuple.ImmutablePair; +import org.apache.commons.lang3.tuple.Pair; import org.apache.fineract.infrastructure.core.config.FineractProperties; import org.apache.fineract.infrastructure.core.messaging.jms.MessageFactory; import org.apache.fineract.infrastructure.core.service.HashingService; @@ -73,8 +76,12 @@ public class JMSMultiExternalEventProducer implements ExternalEventProducer { public void sendEvents(Map> partitions) throws AcknowledgementTimeoutException { Map> indexedPartitions = mapPartitionsToProducers(partitions); measure(() -> { - List> tasks = sendPartitions(indexedPartitions); + List> producersWithSessions = obtainProducers(); + List producers = producersWithSessions.stream().map(Pair::getRight).collect(Collectors.toList()); + List sessions = producersWithSessions.stream().map(Pair::getLeft).collect(Collectors.toList()); + List> tasks = sendPartitions(indexedPartitions, producers); waitForSendingCompletion(tasks); + closeSessions(sessions); }, timeTaken -> { if (log.isDebugEnabled()) { int eventCount = partitions.values().stream().map(Collection::size).reduce(0, Integer::sum); @@ -84,8 +91,20 @@ public class JMSMultiExternalEventProducer implements ExternalEventProducer { }); } - private List obtainProducers() { - List result = new ArrayList<>(); + private void closeSessions(List sessions) { + // The sessions retrieved from a CachingConnectionFactory needs to be explicitly closed, otherwise we're making + // orphan sessions, leaking memory + for (Session session : sessions) { + try { + session.close(); + } catch (JMSException e) { + log.error("Exception while trying to close sessions", e); + } + } + } + + private List> obtainProducers() { + List> result = new ArrayList<>(); int producerCount = getProducerCount(); try { // No need to close the connection since it's a pooled one @@ -96,7 +115,7 @@ public class JMSMultiExternalEventProducer implements ExternalEventProducer { // producers Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); MessageProducer producer = session.createProducer(destination); - result.add(producer); + result.add(new ImmutablePair<>(session, producer)); } } catch (JMSException e) { throw new RuntimeException("Error while obtaining message producers", e); @@ -104,8 +123,7 @@ public class JMSMultiExternalEventProducer implements ExternalEventProducer { return result; } - private List> sendPartitions(Map> indexedPartitions) { - List producers = obtainProducers(); + private List> sendPartitions(Map> indexedPartitions, List producers) { List> tasks = new ArrayList<>(); for (Map.Entry> entry : indexedPartitions.entrySet()) { Integer producerIndex = entry.getKey(); diff --git a/fineract-provider/src/test/java/org/apache/fineract/infrastructure/event/external/producer/jms/JMSMultiExternalEventProducerTest.java b/fineract-provider/src/test/java/org/apache/fineract/infrastructure/event/external/producer/jms/JMSMultiExternalEventProducerTest.java index 8cd0dd431..79fc715fe 100644 --- a/fineract-provider/src/test/java/org/apache/fineract/infrastructure/event/external/producer/jms/JMSMultiExternalEventProducerTest.java +++ b/fineract-provider/src/test/java/org/apache/fineract/infrastructure/event/external/producer/jms/JMSMultiExternalEventProducerTest.java @@ -37,6 +37,7 @@ import org.apache.fineract.avro.MessageV1; import org.apache.fineract.infrastructure.core.config.FineractProperties; import org.apache.fineract.infrastructure.core.messaging.jms.MessageFactory; import org.apache.fineract.infrastructure.core.service.HashingService; +import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; @@ -109,6 +110,13 @@ class JMSMultiExternalEventProducerTest { given(session3.createProducer(destination)).willReturn(producer3); } + @AfterEach + public void tearDown() throws Exception { + verify(session1).close(); + verify(session2).close(); + verify(session3).close(); + } + @Test public void testSendEventsShouldWork() throws Exception { // given