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); + } +}