Unable to publish MQTT message

[ljamesd15]

Hey all I am currently attempting to publish a MQTT message to my RabbitMQ broker but it doesn't seem to be working

I have attempted to publish a message using mosquitto_pub

$ mosquitto_pub -h localhost -t test -m "test" -p 1883
Error: The connection was lost.

and also using a Java application which fails to connect to the broker

Caused by: java.lang.RuntimeException: Connection lost (32109) - java.io.EOFException
	at com.personal.mqtt.mqttexamples.service.MqttService.<init>(MqttService.java:42) ~[classes/:na]
	at java.base/jdk.internal.reflect.DirectConstructorHandleAccessor.newInstance(DirectConstructorHandleAccessor.java:62) ~[na:na]
	at java.base/java.lang.reflect.Constructor.newInstanceWithCaller(Constructor.java:502) ~[na:na]
	at java.base/java.lang.reflect.Constructor.newInstance(Constructor.java:486) ~[na:na]
	at org.springframework.beans.BeanUtils.instantiateClass(BeanUtils.java:197) ~[spring-beans-6.1.3.jar:6.1.3]
	... 33 common frames omitted
Caused by: org.eclipse.paho.client.mqttv3.MqttException: Connection lost
	at org.eclipse.paho.client.mqttv3.internal.CommsReceiver.run(CommsReceiver.java:197) ~[org.eclipse.paho.client.mqttv3-1.2.5.jar:na]
	at java.base/java.lang.Thread.run(Thread.java:1583) ~[na:na]
Caused by: java.io.EOFException: null
	at java.base/java.io.DataInputStream.readUnsignedByte(DataInputStream.java:297) ~[na:na]
	at java.base/java.io.DataInputStream.readByte(DataInputStream.java:275) ~[na:na]
	at org.eclipse.paho.client.mqttv3.internal.wire.MqttInputStream.readMqttWireMessage(MqttInputStream.java:92) ~[org.eclipse.paho.client.mqttv3-1.2.5.jar:na]
	at org.eclipse.paho.client.mqttv3.internal.CommsReceiver.run(CommsReceiver.java:137) ~[org.eclipse.paho.client.mqttv3-1.2.5.jar:na]
	... 1 common frames omitted

Java code snippet is

        try {
            this.publisher = new MqttClient("tcp://127.0.0.1:1883", MqttClient.generateClientId(), new MemoryPersistence());
        } catch (MqttException e) {
            throw new RuntimeException(e);
        }
        final MqttConnectOptions options = new MqttConnectOptions();
        options.setAutomaticReconnect(true);
        options.setKeepAliveInterval(100);
        options.setCleanSession(false);
        options.setConnectionTimeout(10);

        try {
            publisher.connect(options); <---- Fails here
        } catch (MqttException e) {
            throw new RuntimeException(e);
        }

and from the RabbitMQ logs I see

2024-02-19 18:02:57.675268-08:00 [info] <0.1742.0> accepting MQTT connection <0.1742.0> (127.0.0.1:53966 -> 127.0.0.1:1883, client id: Phc5klp86hmvv0v2E1p3pA)
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0> ** Generic server <0.1742.0> terminating
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0> ** Last message in was {'EXIT',#Port<0.357>,einval}
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0> ** When Server state == {state,#Port<0.357>,
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                          "127.0.0.1:53966 -> 127.0.0.1:1883",true,undefined,
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                          true,running,
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                          {none,<0.1767.0>},
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                          <0.1741.0>,false,none,
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                          {proc_state,#Port<0.357>,#{},
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                           {undefined,undefined},
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                           {0,nil},
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                           {0,nil},
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                           undefined,1,"Phc5klp86hmvv0v2E1p3pA",true,undefined,
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                           {<0.1753.0>,undefined},
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                           <0.1745.0>,<<"amq.topic">>,
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                           {amqp_adapter_info,
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                            {0,0,0,0,0,65535,32512,1},
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                            1883,
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                            {0,0,0,0,0,65535,32512,1},
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                            53966,<<"127.0.0.1:53966 -> 127.0.0.1:1883">>,
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                            {'MQTT',"N/A"},
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                            [{variable_map,
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                              #{<<"client_id">> =>
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                                 <<"Phc5klp86hmvv0v2E1p3pA">>}},
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                             {channels,1},
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                             {channel_max,1},
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                             {frame_max,0},
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                             {client_properties,
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                              [{<<"product">>,longstr,<<"MQTT client">>},
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                               {client_id,longstr,
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                                <<"Phc5klp86hmvv0v2E1p3pA">>}]},
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                             {ssl,false}]},
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                           none,<0.645.0>,
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                           {auth_state,<<"guest">>,
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                            {user,<<"guest">>,
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                             [administrator],
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                             [{rabbit_auth_backend_internal,none}]},
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                            <<"weather">>},
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                           #Fun<rabbit_mqtt_processor.0.58947706>,
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                           {0,0,0,0,0,65535,32512,1},
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                           #Fun<rabbit_mqtt_util.4.50356794>,
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                           #Fun<rabbit_mqtt_util.5.50356794>,
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                           {pending,#Ref<0.3022418089.1311768577.41103>}},
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                          <0.1745.0>,
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                          {state,fine,5000,
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0>                           #Ref<0.3022418089.1311768577.41183>}}
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0> ** Reason for termination ==
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0> ** {connection_died,einval}
2024-02-19 18:02:57.675553-08:00 [error] <0.1742.0> 
2024-02-19 18:02:57.676361-08:00 [error] <0.1742.0>   crasher:
2024-02-19 18:02:57.676361-08:00 [error] <0.1742.0>     initial call: rabbit_mqtt_reader:init/1
2024-02-19 18:02:57.676361-08:00 [error] <0.1742.0>     pid: <0.1742.0>
2024-02-19 18:02:57.676361-08:00 [error] <0.1742.0>     registered_name: []
2024-02-19 18:02:57.676361-08:00 [error] <0.1742.0>     exception exit: {connection_died,einval}
2024-02-19 18:02:57.676361-08:00 [error] <0.1742.0>       in function  gen_server2:terminate/3 (gen_server2.erl, line 1183)
2024-02-19 18:02:57.676361-08:00 [error] <0.1742.0>     ancestors: [<0.1740.0>,<0.653.0>,<0.652.0>,<0.651.0>,<0.649.0>,
2024-02-19 18:02:57.676361-08:00 [error] <0.1742.0>                   <0.648.0>,rabbit_mqtt_sup,<0.636.0>]
2024-02-19 18:02:57.676361-08:00 [error] <0.1742.0>     message_queue_len: 2
2024-02-19 18:02:57.676361-08:00 [error] <0.1742.0>     messages: [{'EXIT',<0.1753.0>,normal},
2024-02-19 18:02:57.676361-08:00 [error] <0.1742.0>                   {'EXIT',<0.1745.0>,{shutdown,normal}}]
2024-02-19 18:02:57.676361-08:00 [error] <0.1742.0>     links: [<0.1740.0>]
2024-02-19 18:02:57.676361-08:00 [error] <0.1742.0>     dictionary: [{rand_seed,{#{max => 288230376151711743,type => exsplus,
2024-02-19 18:02:57.676361-08:00 [error] <0.1742.0>                                 next => #Fun<rand.5.65977474>,
2024-02-19 18:02:57.676361-08:00 [error] <0.1742.0>                                 jump => #Fun<rand.3.65977474>},
2024-02-19 18:02:57.676361-08:00 [error] <0.1742.0>                               [259312428527719485|245068100575567147]}},
2024-02-19 18:02:57.676361-08:00 [error] <0.1742.0>                   {gen_server_call_timeout,130000},
2024-02-19 18:02:57.676361-08:00 [error] <0.1742.0>                   {guid_secure,{{10,'rabbit@littleton-ubuntu-5566',
2024-02-19 18:02:57.676361-08:00 [error] <0.1742.0>                                  #Ref<0.3022418089.1311768577.41066>},
2024-02-19 18:02:57.676361-08:00 [error] <0.1742.0>                                 0}}]
2024-02-19 18:02:57.676361-08:00 [error] <0.1742.0>     trap_exit: true
2024-02-19 18:02:57.676361-08:00 [error] <0.1742.0>     status: running
2024-02-19 18:02:57.676361-08:00 [error] <0.1742.0>     heap_size: 4185
2024-02-19 18:02:57.676361-08:00 [error] <0.1742.0>     stack_size: 28
2024-02-19 18:02:57.676361-08:00 [error] <0.1742.0>     reductions: 34590
2024-02-19 18:02:57.676361-08:00 [error] <0.1742.0>   neighbours:
2024-02-19 18:02:57.676361-08:00 [error] <0.1742.0> 
2024-02-19 18:02:57.677030-08:00 [error] <0.1740.0>     supervisor: {<0.1740.0>,rabbit_mqtt_connection_sup}
2024-02-19 18:02:57.677030-08:00 [error] <0.1740.0>     errorContext: child_terminated
2024-02-19 18:02:57.677030-08:00 [error] <0.1740.0>     reason: {connection_died,einval}
2024-02-19 18:02:57.677030-08:00 [error] <0.1740.0>     offender: [{pid,<0.1742.0>},
2024-02-19 18:02:57.677030-08:00 [error] <0.1740.0>                {id,rabbit_mqtt_reader},
2024-02-19 18:02:57.677030-08:00 [error] <0.1740.0>                {mfargs,
2024-02-19 18:02:57.677030-08:00 [error] <0.1740.0>                    {rabbit_mqtt_reader,start_link,
2024-02-19 18:02:57.677030-08:00 [error] <0.1740.0>                        [<0.1741.0>,{acceptor,{0,0,0,0,0,0,0,0},1883}]}},
2024-02-19 18:02:57.677030-08:00 [error] <0.1740.0>                {restart_type,intrinsic},
2024-02-19 18:02:57.677030-08:00 [error] <0.1740.0>                {shutdown,300000},
2024-02-19 18:02:57.677030-08:00 [error] <0.1740.0>                {child_type,worker}]
2024-02-19 18:02:57.677202-08:00 [error] <0.1740.0>     supervisor: {<0.1740.0>,rabbit_mqtt_connection_sup}
2024-02-19 18:02:57.677202-08:00 [error] <0.1740.0>     errorContext: shutdown
2024-02-19 18:02:57.677202-08:00 [error] <0.1740.0>     reason: reached_max_restart_intensity
2024-02-19 18:02:57.677202-08:00 [error] <0.1740.0>     offender: [{pid,<0.1742.0>},
2024-02-19 18:02:57.677202-08:00 [error] <0.1740.0>                {id,rabbit_mqtt_reader},
2024-02-19 18:02:57.677202-08:00 [error] <0.1740.0>                {mfargs,
2024-02-19 18:02:57.677202-08:00 [error] <0.1740.0>                    {rabbit_mqtt_reader,start_link,
2024-02-19 18:02:57.677202-08:00 [error] <0.1740.0>                        [<0.1741.0>,{acceptor,{0,0,0,0,0,0,0,0},1883}]}},
2024-02-19 18:02:57.677202-08:00 [error] <0.1740.0>                {restart_type,intrinsic},
2024-02-19 18:02:57.677202-08:00 [error] <0.1740.0>                {shutdown,300000},
2024-02-19 18:02:57.677202-08:00 [error] <0.1740.0>                {child_type,worker}]

My RabbitMQ node appears all fine,

$ sudo rabbitmqctl status
Runtime

OS PID: 224758
OS: Linux
Uptime (seconds): 1502
Is under maintenance?: false
RabbitMQ version: 3.9.13
Node name: rabbit@littleton-ubuntu-5566
Erlang configuration: Erlang/OTP 26 [erts-14.2.2] [source] [64-bit] [smp:4:4] [ds:4:4:10] [async-threads:1] [jit:ns]
Erlang processes: 416 used, 1048576 limit
Scheduler run queue: 1
Cluster heartbeat timeout (net_ticktime): 60

Plugins

Enabled plugin file: /etc/rabbitmq/enabled_plugins
Enabled plugins:

 * rabbitmq_mqtt
 * rabbitmq_management
 * amqp_client
 * rabbitmq_web_dispatch
 * cowboy
 * cowlib
 * rabbitmq_management_agent

Data directory

Node data directory: /var/lib/rabbitmq/mnesia/rabbit@littleton-ubuntu-5566
Raft data directory: /var/lib/rabbitmq/mnesia/rabbit@littleton-ubuntu-5566/quorum/rabbit@littleton-ubuntu-5566

Config files

 * /etc/rabbitmq/rabbitmq.conf

Log file(s)

 * /var/log/rabbitmq/[email protected]
 * /var/log/rabbitmq/rabbit@littleton-ubuntu-5566_upgrade.log
 * <stdout>

Alarms

(none)

Memory

Total memory used: 0.1536 gb
Calculation strategy: rss
Memory high watermark setting: 0.4 of available memory, computed to: 3.2845 gb

reserved_unallocated: 0.0795 gb (51.76 %)
code: 0.0361 gb (23.5 %)
other_proc: 0.0206 gb (13.42 %)
other_system: 0.02 gb (13.02 %)
other_ets: 0.0029 gb (1.88 %)
atom: 0.0015 gb (0.96 %)
plugins: 0.001 gb (0.66 %)
metrics: 0.0007 gb (0.48 %)
binary: 0.0005 gb (0.34 %)
mgmt_db: 0.0002 gb (0.14 %)
mnesia: 0.0001 gb (0.07 %)
quorum_ets: 0.0001 gb (0.07 %)
queue_procs: 0.0001 gb (0.06 %)
msg_index: 0.0001 gb (0.04 %)
connection_other: 0.0001 gb (0.03 %)
stream_queue_procs: 0.0 gb (0.0 %)
stream_queue_replica_reader_procs: 0.0 gb (0.0 %)
connection_readers: 0.0 gb (0.0 %)
connection_writers: 0.0 gb (0.0 %)
connection_channels: 0.0 gb (0.0 %)
queue_slave_procs: 0.0 gb (0.0 %)
quorum_queue_procs: 0.0 gb (0.0 %)
stream_queue_coordinator_procs: 0.0 gb (0.0 %)
allocated_unused: 0.0 gb (0.0 %)

File Descriptors

Total: 6, limit: 927
Sockets: 0, limit: 832

Free Disk Space

Low free disk space watermark: 0.05 gb
Free disk space: 112.2944 gb

Totals

Connection count: 0
Queue count: 3
Virtual host count: 2

Listeners

Interface: [::], port: 15672, protocol: http, purpose: HTTP API
Interface: [::], port: 1883, protocol: mqtt, purpose: MQTT
Interface: [::], port: 25672, protocol: clustering, purpose: inter-node and CLI tool communication
Interface: [::], port: 5672, protocol: amqp, purpose: AMQP 0-9-1 and AMQP 1.0

See my RabbitMQ conf

$ sudo cat /etc/rabbitmq/rabbitmq.conf
mqtt.listeners.tcp.default = 1883
## Default MQTT with TLS port is 8883
# mqtt.listeners.ssl.default = 8883

mqtt.allow_anonymous  = true
mqtt.default_user     = guest
mqtt.default_pass     = guest

mqtt.vhost            = weather
# 24 hours by default
mqtt.subscription_ttl = 86400000
mqtt.prefetch         = 10

In the RabbitMQ management UI I am no where near exhausting resources everything is under 10% usage. Given that its just a single message I'm assuming there is a setting I have incorrect but the RabbitMQ logs are strange because they seem to mention "reached_max_restart_intensity" which after googling seems to be a common problem under load

Any help is appreciated! Let me know if I can provide any other info.

[michaelklishin]

I'm afraid the version you run, RabbitMQ, 3.9, has reached end of life.

#9529 and #2730 are the most relevant issues. einval is a fairly general socket error message that does not really provide much details what specifically goes wrong during the connection acceptance step.

But EOL means EOL. Please give 3.12.13 a shot.