### Question:
MQTTStreamSource extends BlockingQueueStreamSource<List<Object>> { @Override public void open() { try { mqttClient = MQTTClientBuilder.builder() .withBroker( mqttConsumerMeta.getMqttServer() ) .withTopics( mqttConsumerMeta.getTopics() ) .withQos( mqttConsumerMeta.getQos() ) .withStep( mqttConsumer ) .withCallback( callback ) .withUsername( mqttConsumerMeta.getUsername() ) .withPassword( mqttConsumerMeta.getPassword() ) .withIsSecure( mqttConsumerMeta.isUseSsl() ) .withSslConfig( mqttConsumerMeta.getSslConfig() ) .withKeepAliveInterval( mqttConsumerMeta.getKeepAliveInterval() ) .withMaxInflight( mqttConsumerMeta.getMaxInflight() ) .withConnectionTimeout( mqttConsumerMeta.getConnectionTimeout() ) .withCleanSession( mqttConsumerMeta.getCleanSession() ) .withStorageLevel( mqttConsumerMeta.getStorageLevel() ) .withServerUris( mqttConsumerMeta.getServerUris() ) .withMqttVersion( mqttConsumerMeta.getMqttVersion() ) .withAutomaticReconnect( mqttConsumerMeta.getAutomaticReconnect() ) .buildAndConnect(); } catch ( Exception e ) { mqttConsumer.stopAll(); mqttConsumer.logError( e.toString() ); } } MQTTStreamSource( MQTTConsumerMeta mqttConsumerMeta, MQTTConsumer mqttConsumer ); @Override void open(); @Override void close();  }### Answer:
@Test public void testMQTTOpenException() throws Exception { PowerMockito.mockStatic( MQTTClientBuilder.class ); MQTTClientBuilder.ClientFactory clientFactory = mock( MQTTClientBuilder.ClientFactory.class ); MqttClient mqttClient = mock( MqttClient.class ); MQTTClientBuilder builder = spy( MQTTClientBuilder.class ); MqttException mqttException = mock( MqttException.class ); when( clientFactory.getClient( any(), any(), any() ) ).thenReturn( mqttClient ); when( mqttException.toString() ).thenReturn( "There is an error connecting" ); doThrow( mqttException ).when( builder ).buildAndConnect(); PowerMockito.when( MQTTClientBuilder.builder() ).thenReturn( builder ); MQTTStreamSource source = new MQTTStreamSource( consumerMeta, mqttConsumer ); source.open(); verify( mqttConsumer ).stopAll(); verify( mqttConsumer ).logError( "There is an error connecting" ); }