From a0d39c40510e3a6a6dedf8fcc5bb03e54a5428ef Mon Sep 17 00:00:00 2001 From: Craig <3979063+craig8@users.noreply.github.com> Date: Fri, 10 Jul 2026 11:00:10 -0700 Subject: [PATCH] fix: close subscription consumers on GossClient.close() to stop resource leak subscribe() created a DefaultClientConsumer but never stored it, so the JMS consumer and its listener thread were abandoned, never deregistered from the broker. Track subscription consumers and close each on client close() before the session closes. Each close() is individually guarded so one throwing consumer cannot strand the remaining consumers or the session/connection teardown. DefaultClientConsumer.close() guards a null messageConsumer (a partially-constructed consumer) and logs a close failure at error instead of printing the stack trace. Tests cover tracking, teardown, and the throwing consumer error path. --- .../core/client/DefaultClientConsumer.java | 10 +- .../src/pnnl/goss/core/client/GossClient.java | 32 ++++- .../test/GossClientConsumerLeakTest.java | 134 ++++++++++++++++++ 3 files changed, 171 insertions(+), 5 deletions(-) create mode 100644 pnnl.goss.core/test/pnnl/goss/core/client/test/GossClientConsumerLeakTest.java diff --git a/pnnl.goss.core/src/pnnl/goss/core/client/DefaultClientConsumer.java b/pnnl.goss.core/src/pnnl/goss/core/client/DefaultClientConsumer.java index 6cd4ebba..3265fa56 100644 --- a/pnnl.goss.core/src/pnnl/goss/core/client/DefaultClientConsumer.java +++ b/pnnl.goss.core/src/pnnl/goss/core/client/DefaultClientConsumer.java @@ -81,10 +81,18 @@ public DefaultClientConsumer(Session session, Destination destination) { } public void close() { + // messageConsumer can be null when the constructor's session.createConsumer() + // call failed: the constructor logs and swallows the exception rather than + // rethrowing, so a partially-constructed consumer with a null messageConsumer + // can still end up tracked and later closed. Guard against that null so close() + // cannot NPE on a consumer that never finished construction. + if (getMessageConsumer() == null) { + return; + } try { getMessageConsumer().close(); } catch (JMSException e) { - e.printStackTrace(); + log.error("Failed to close message consumer", e); } } diff --git a/pnnl.goss.core/src/pnnl/goss/core/client/GossClient.java b/pnnl.goss.core/src/pnnl/goss/core/client/GossClient.java index 3bc1046e..2b730fb9 100644 --- a/pnnl.goss.core/src/pnnl/goss/core/client/GossClient.java +++ b/pnnl.goss.core/src/pnnl/goss/core/client/GossClient.java @@ -104,6 +104,11 @@ public class GossClient implements Client { private String trustStore; private String trustStorePassword; private List threads = new ArrayList(); + // Consumers created by subscribe() are long-lived (unlike the getResponse() + // consumer, which is closed in its own try/finally). Track them here so + // close() can deregister every subscription's MessageConsumer from the + // broker instead of leaking it when the reference goes out of scope. + private final List subscriptionConsumers = new ArrayList(); private PROTOCOL protocol; private Credentials credentials = null; private String token = null; @@ -366,8 +371,9 @@ public Client subscribe(String topicName, GossResponseEvent event) if (this.protocol.equals(PROTOCOL.OPENWIRE) || this.protocol.equals(PROTOCOL.STOMP)) { // Both OPENWIRE and STOMP use the same JMS patterns with ActiveMQ destination = getDestination(topicName); - new DefaultClientConsumer(new DefaultClientListener(event), - session, destination); + ClientConsumer clientConsumer = new DefaultClientConsumer( + new DefaultClientListener(event), session, destination); + subscriptionConsumers.add(clientConsumer); } } finally { @@ -399,8 +405,9 @@ public Client subscribe(String destinationName, GossResponseEvent event, DESTINA if (this.protocol.equals(PROTOCOL.OPENWIRE) || this.protocol.equals(PROTOCOL.STOMP)) { // Both OPENWIRE and STOMP use the same JMS patterns with ActiveMQ destination = getDestination(destinationName, destinationType); - new DefaultClientConsumer(new DefaultClientListener(event), - session, destination); + ClientConsumer clientConsumer = new DefaultClientConsumer( + new DefaultClientListener(event), session, destination); + subscriptionConsumers.add(clientConsumer); } } finally { @@ -603,6 +610,23 @@ public void publish(Destination destination, Serializable data) throws SystemExc public void close() { try { log.debug("Client closing!"); + + // Close every consumer created by subscribe() before the session closes, + // so the MessageConsumer (and its listener thread) is deregistered from + // the broker instead of being abandoned. Each close() is individually + // guarded: one consumer throwing must not skip the remaining consumers, + // the list clear below, or the session/connection teardown that follows, + // otherwise a single bad consumer strands the authenticated session. + for (ClientConsumer clientConsumer : subscriptionConsumers) { + try { + clientConsumer.close(); + } catch (Exception e) { + log.warn("Failed to close subscription consumer {}; continuing to close remaining consumers", + clientConsumer, e); + } + } + subscriptionConsumers.clear(); + if (session != null) { session.close(); session = null; diff --git a/pnnl.goss.core/test/pnnl/goss/core/client/test/GossClientConsumerLeakTest.java b/pnnl.goss.core/test/pnnl/goss/core/client/test/GossClientConsumerLeakTest.java new file mode 100644 index 00000000..e6f0869a --- /dev/null +++ b/pnnl.goss.core/test/pnnl/goss/core/client/test/GossClientConsumerLeakTest.java @@ -0,0 +1,134 @@ +package pnnl.goss.core.client.test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.lang.reflect.Field; +import java.util.ArrayList; +import java.util.List; + +import jakarta.jms.MessageConsumer; +import jakarta.jms.Session; +import jakarta.jms.Topic; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; + +import pnnl.goss.core.Client.PROTOCOL; +import pnnl.goss.core.ClientConsumer; +import pnnl.goss.core.client.GossClient; + +/** + * Regression coverage for GADP-018: GossClient.subscribe() created a + * DefaultClientConsumer and dropped the reference immediately after the method + * returned, so the underlying JMS MessageConsumer (and its listener thread) was + * never closed and stayed registered on the broker for the lifetime of the + * connection. + * + * GossClient has no broker-backed constructor path suitable for a unit test + * (createSession() dials a real ActiveMQConnectionFactory), so these tests + * inject a mocked Session via reflection to exercise subscribe() and close() + * without a live broker. This covers the unit-level contract (consumer tracked, + * closed, list cleared); it does not cover the broker-observable consequence + * (the consumer count on the ActiveMQ destination actually drops to zero), + * which needs a live-broker integration test. + */ +public class GossClientConsumerLeakTest { + + private GossClient client; + private Session mockSession; + private MessageConsumer mockConsumer; + + @BeforeEach + void setUp() throws Exception { + client = new GossClient(PROTOCOL.OPENWIRE, null, "tcp://localhost:61616", + "stomp://localhost:61613"); + + mockSession = mock(Session.class); + Topic mockTopic = mock(Topic.class); + mockConsumer = mock(MessageConsumer.class); + when(mockSession.createTopic("test.topic")).thenReturn(mockTopic); + when(mockSession.createConsumer(mockTopic)).thenReturn(mockConsumer); + + setPrivateField(client, "session", mockSession); + } + + @Test + @DisplayName("subscribe() tracks the created consumer instead of dropping the reference") + void subscribeTracksConsumerForLifecycleManagement() throws Exception { + client.subscribe("test.topic", response -> { + }); + + List tracked = getPrivateField(client, "subscriptionConsumers", List.class); + assertThat(tracked).hasSize(1); + assertThat(tracked.get(0)).isInstanceOf(ClientConsumer.class); + } + + @Test + @DisplayName("close() closes every MessageConsumer created by subscribe(), deregistering it from the broker") + void closeClosesTrackedConsumers() throws Exception { + client.subscribe("test.topic", response -> { + }); + + client.close(); + + verify(mockConsumer, times(1)).close(); + } + + @Test + @DisplayName("close() clears the tracked-consumer list so a second close() is a no-op, not a double-close") + void closeClearsTrackedConsumers() throws Exception { + client.subscribe("test.topic", response -> { + }); + client.close(); + + List tracked = getPrivateField(client, "subscriptionConsumers", List.class); + assertThat(tracked).isEmpty(); + } + + @Test + @DisplayName("close() tolerates a consumer whose close() throws: remaining consumers still close, " + + "the tracked list still clears, and session teardown still runs") + void closeSurvivesAThrowingConsumerAndStillTearsDownSession() throws Exception { + // Regression guard for the HIGH finding: a consumer left in a bad state + // (e.g. a null messageConsumer from a failed subscribe()) must not abort + // the close() loop and strand the remaining consumers, the tracked list, + // or the session/connection teardown that follows. + ClientConsumer throwingConsumer = mock(ClientConsumer.class); + doThrow(new RuntimeException("boom")).when(throwingConsumer).close(); + ClientConsumer healthyConsumer = mock(ClientConsumer.class); + + List tracked = new ArrayList<>(); + tracked.add(throwingConsumer); + tracked.add(healthyConsumer); + setPrivateField(client, "subscriptionConsumers", tracked); + + client.close(); + + verify(throwingConsumer, times(1)).close(); + verify(healthyConsumer, times(1)).close(); + + List trackedAfter = getPrivateField(client, "subscriptionConsumers", List.class); + assertThat(trackedAfter).isEmpty(); + + verify(mockSession, times(1)).close(); + } + + @SuppressWarnings("unchecked") + private static T getPrivateField(Object target, String name, Class type) throws Exception { + Field field = target.getClass().getDeclaredField(name); + field.setAccessible(true); + return (T) field.get(target); + } + + private static void setPrivateField(Object target, String name, Object value) throws Exception { + Field field = target.getClass().getDeclaredField(name); + field.setAccessible(true); + field.set(target, value); + } +}