FINERACT-1724: Closing JMS sessions explicitly to avoid memory leaking for multi external event producer

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<Long, List<byte[]>> partitions) throws AcknowledgementTimeoutException {
         Map<Integer, List<byte[]>> indexedPartitions = mapPartitionsToProducers(partitions);
         measure(() -> {
-            List<Future<?>> tasks = sendPartitions(indexedPartitions);
+            List<Pair<Session, MessageProducer>> producersWithSessions = obtainProducers();
+            List<MessageProducer> producers = producersWithSessions.stream().map(Pair::getRight).collect(Collectors.toList());
+            List<Session> sessions = producersWithSessions.stream().map(Pair::getLeft).collect(Collectors.toList());
+            List<Future<?>> 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<MessageProducer> obtainProducers() {
+    private void closeSessions(List<Session> sessions) {
-        List<MessageProducer> result = new ArrayList<>();
+        // 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<Pair<Session, MessageProducer>> obtainProducers() {
+        List<Pair<Session, MessageProducer>> 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<Future<?>> sendPartitions(Map<Integer, List<byte[]>> indexedPartitions) {
+    private List<Future<?>> sendPartitions(Map<Integer, List<byte[]>> indexedPartitions, List<MessageProducer> producers) {
-        List<MessageProducer> producers = obtainProducers();
         List<Future<?>> tasks = new ArrayList<>();
         for (Map.Entry<Integer, List<byte[]>> entry : indexedPartitions.entrySet()) {
             Integer producerIndex = entry.getKey();

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