JMSMessageConsumer { void close() { try { if (session != null) { session.close(); } } catch (JMSException e) { logger.error("Could not destroy session", e); } try { if (connection != null) { connection.close(); } } catch (JMSException e) { logger.error("Could not destroy connection", e); } } JMSMessageConsumer(InitialContext initialContext, ConnectionFactory connectionFactory,
                     String destinationName, JMSDestinationLocator destinationLocator,
                     JMSDestinationType destinationType, String messageSelector, int batchSize,
                     long pollTimeout, JMSMessageConverter messageConverter,
                     Optional<String> userName, Optional<String> password,
                     Optional<String> clientId, boolean createDurableSubscription,
                     String durableSubscriptionName);   }
@Test public void testCreateSessionFails() throws Exception { when(connection.createSession(true, Session.SESSION_TRANSACTED)) .thenThrow(new JMSException("")); try { create(); fail("Expected exception: org.apache.flume.FlumeException"); } catch (FlumeException e) { verify(connection).close(); } }
@Test public void testCreateQueueFails() throws Exception { when(session.createQueue(destinationName)) .thenThrow(new JMSException("")); try { create(); fail("Expected exception: org.apache.flume.FlumeException"); } catch (FlumeException e) { verify(session).close(); verify(connection).close(); } }
@Test public void testCreateTopicFails() throws Exception { destinationType = JMSDestinationType.TOPIC; when(session.createTopic(destinationName)) .thenThrow(new JMSException("")); try { create(); fail("Expected exception: org.apache.flume.FlumeException"); } catch (FlumeException e) { verify(session).close(); verify(connection).close(); } }
@Test public void testCreateConsumerFails() throws Exception { when(session.createConsumer(any(Destination.class), anyString())) .thenThrow(new JMSException("")); try { create(); fail("Expected exception: org.apache.flume.FlumeException"); } catch (FlumeException e) { verify(session).close(); verify(connection).close(); } }
@Test public void testClose() throws Exception { doThrow(new JMSException("")).when(session).close(); consumer = create(); consumer.close(); verify(session, times(1)).close(); verify(connection, times(1)).close(); }