Support Questions

Find answers, ask questions, and share your expertise
Check out our newest addition to the community, the Cloudera Data Analytics (CDA) group hub.

NiFi 1.4 ConsumeMQTT Processor throws an error - MqttException (128)

Hi everyone,

we are using NiFi 1.4 and the ConsumeMQTT processor to subscribe to topics publish by a message broker from Web Methods calls Universal Messaging.

For some channels we experience unexpected behavior.

When I start the ConsumeMQTT Processor I get the following error message:

2018-08-29 09:05:16,615 ERROR [Timer-Driven Process Thread-1] o.a.nifi.processors.mqtt.ConsumeMQTT ConsumeMQTT[id=01651015-56c3-1f18-9c66-35570bec2f7a] Connection to tcp:// lost (or was never connected) and ontrigger connect failed. Yielding processor:  
org.eclipse.paho.client.mqttv3.MqttException: MqttException
 at org.eclipse.paho.client.mqttv3.MqttClient.subscribe(
 at org.eclipse.paho.client.mqttv3.MqttClient.subscribe(
 at org.apache.nifi.processors.mqtt.ConsumeMQTT.reconnect(
 at org.apache.nifi.processors.mqtt.ConsumeMQTT.onTrigger(
 at org.apache.nifi.processors.mqtt.common.AbstractMQTTProcessor.onTrigger(
 at org.apache.nifi.controller.StandardProcessorNode.onTrigger(
 at org.apache.nifi.controller.scheduling.TimerDrivenSchedulingAgent$
 at java.util.concurrent.Executors$
 at java.util.concurrent.FutureTask.runAndReset(
 at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(
 at java.util.concurrent.ScheduledThreadPoolExecutor$
 at java.util.concurrent.ThreadPoolExecutor.runWorker(
 at java.util.concurrent.ThreadPoolExecutor$

Attached a screenshot with the processor settings.


Error is only thrown one time and the people who administer the broker told us they are not able to see the connection.

We are not getting the messages.

If I have to guess I would say the problem is related to the QoS value, but I have no clues.

Any idea or comment would be appreciated.

Kind regards,




@Paul Hernandez Are you using nifi in cluster mode? If yes, you can try change consume MQTT to primary node instead of all nodes. Hope it helps!

Cloudera Employee

It worked. Thanks!

Take a Tour of the Community
Don't have an account?
Your experience may be limited. Sign in to explore more.