@Test public void testStartConsumerCreateThrowsException() throws Exception { doThrow(new RuntimeException("Expected")).when(source).createConsumer(); source.configure(context); source.start(); try { source.process(); Assert.fail(); } catch (FlumeException expected) { } }
@VisibleForTesting JMSMessageConsumer createConsumer() throws JMSException { logger.info("Creating new consumer for " + destinationName); JMSMessageConsumer consumer = new JMSMessageConsumer(initialContext, connectionFactory, destinationName, destinationLocator, destinationType, messageSelector, batchSize, pollTimeout, converter, userName, password, clientId, createDurableSubscription, durableSubscriptionName); jmsExceptionCounter = 0; return consumer; }
JMSSource extends AbstractPollableSource implements BatchSizeSupported { @VisibleForTesting JMSMessageConsumer createConsumer() throws JMSException { logger.info("Creating new consumer for " + destinationName); JMSMessageConsumer consumer = new JMSMessageConsumer(initialContext, connectionFactory, destinationName, destinationLocator, destinationType, messageSelector, batchSize, pollTimeout, converter, userName, password, clientId, createDurableSubscription, durableSubscriptionName); jmsExceptionCounter = 0; return consumer; } }
JMSSource extends AbstractPollableSource implements BatchSizeSupported { @VisibleForTesting JMSMessageConsumer createConsumer() throws JMSException { logger.info("Creating new consumer for " + destinationName); JMSMessageConsumer consumer = new JMSMessageConsumer(initialContext, connectionFactory, destinationName, destinationLocator, destinationType, messageSelector, batchSize, pollTimeout, converter, userName, password, clientId, createDurableSubscription, durableSubscriptionName); jmsExceptionCounter = 0; return consumer; } JMSSource();  JMSSource(InitialContextFactory initialContextFactory); }
JMSSource extends AbstractPollableSource implements BatchSizeSupported { @VisibleForTesting JMSMessageConsumer createConsumer() throws JMSException { logger.info("Creating new consumer for " + destinationName); JMSMessageConsumer consumer = new JMSMessageConsumer(initialContext, connectionFactory, destinationName, destinationLocator, destinationType, messageSelector, batchSize, pollTimeout, converter, userName, password, clientId, createDurableSubscription, durableSubscriptionName); jmsExceptionCounter = 0; return consumer; } JMSSource();  JMSSource(InitialContextFactory initialContextFactory); @Override long getBatchSize(); }
JMSSource extends AbstractPollableSource implements BatchSizeSupported { @VisibleForTesting JMSMessageConsumer createConsumer() throws JMSException { logger.info("Creating new consumer for " + destinationName); JMSMessageConsumer consumer = new JMSMessageConsumer(initialContext, connectionFactory, destinationName, destinationLocator, destinationType, messageSelector, batchSize, pollTimeout, converter, userName, password, clientId, createDurableSubscription, durableSubscriptionName); jmsExceptionCounter = 0; return consumer; } JMSSource();  JMSSource(InitialContextFactory initialContextFactory); @Override long getBatchSize();  }