@Override public void open() {
    try {
      mqttClient = MQTTClientBuilder.builder()
        .withClientId( mqttConsumerMeta.getClientId() )
        .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() );
    }
  }
@Test
  public void testMQTTOpenException() throws Exception {
    try ( MockedStatic<MQTTClientBuilder> mqttClientBuilderMockedStatic = Mockito.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( mqttException.toString() ).thenReturn( "There is an error connecting" );
      doThrow( mqttException ).when( builder ).buildAndConnect();
      mqttClientBuilderMockedStatic.when( MQTTClientBuilder::builder ).thenReturn( builder );

      MQTTStreamSource source = new MQTTStreamSource( consumerMeta, mqttConsumer );
      source.open();

      verify( mqttConsumer ).stopAll();
      verify( mqttConsumer ).logError( "There is an error connecting" );
    }
  }