Panic on Message Ack when MQTT Client Connection is Lost(QoS 2 + Persistent Session)

[lnxpgn]

Affected Version: v1.5.0 (as used in Telegraf 1.35.4)

We are observing a panic in the paho.mqtt.golang library when an MQTT client loses its network connection and attempts to acknowledge a delivered message after the connection has been lost - specifically when QoS 2 is enabled and a persistent session is active.

The panic happens in the onDelivered function of the mqtt_consumer input plugin of Telegraf, which internally calls msg.Ack() when the persistent session is enabled and track.Delivered() is true — particularly when QoS 2 is used. If the network drops before the acknowledgment can be sent, the client attempts to send the acknowledgment on a closed channel, triggering the panic.

Logs from Telegraf

Sep  2 17:52:21 tvikgge telegraf: 2025-09-02T09:52:21Z E! [inputs.mqtt_consumer] Error in plugin: connection lost: EOF
Sep  2 17:52:21 tvikgge telegraf: panic: send on closed channel
Sep  2 17:52:21 tvikgge telegraf: goroutine 52 [running]:
Sep  2 17:52:21 tvikgge telegraf: github.com/eclipse/paho%2emqtt%2egolang.(*router).matchAndDispatch.func2.ackFunc.4()
Sep  2 17:52:21 tvikgge telegraf: /home/dev/go/pkg/mod/github.com/eclipse/paho.mqtt.golang@v1.5.0/net.go:464 +0x28e
Sep  2 17:52:21 tvikgge telegraf: sync.(*Once).doSlow(0x0?, 0x56280?)
Sep  2 17:52:21 tvikgge telegraf: /home/dev/go-pkg/src/sync/once.go:78 +0xab
Sep  2 17:52:21 tvikgge telegraf: sync.(*Once).Do(...)
Sep  2 17:52:21 tvikgge telegraf: /home/dev/go-pkg/src/sync/once.go:69
Sep  2 17:52:21 tvikgge telegraf: github.com/eclipse/paho%2emqtt%2egolang.(*message).Ack(0x1047e80?)
Sep  2 17:52:21 tvikgge telegraf: /home/dev/go/pkg/mod/github.com/eclipse/paho.mqtt.golang@v1.5.0/message.go:77 +0x25
Sep  2 17:52:21 tvikgge telegraf: github.com/influxdata/telegraf/plugins/inputs/mqtt_consumer.(*MQTTConsumer).onDelivered(0xc0000ca508, {0x13cf190, 0xc000143c80})
Sep  2 17:52:21 tvikgge telegraf: /home/dev/telegraf-1.35.4/plugins/inputs/mqtt_consumer/mqtt_consumer.go:240 +0x12d
Sep  2 17:52:21 tvikgge telegraf: github.com/influxdata/telegraf/plugins/inputs/mqtt_consumer.(*MQTTConsumer).Start.func1()
Sep  2 17:52:21 tvikgge telegraf: /home/dev/telegraf-1.35.4/plugins/inputs/mqtt_consumer/mqtt_consumer.go:145 +0x79
Sep  2 17:52:21 tvikgge telegraf: created by github.com/influxdata/telegraf/plugins/inputs/mqtt_consumer.(*MQTTConsumer).Start in goroutine 1
Sep  2 17:52:21 tvikgge telegraf: /home/dev/telegraf-1.35.4/plugins/inputs/mqtt_consumer/mqtt_consumer.go:138 +0x147
Sep  2 17:52:21 tvikgge systemd: telegraf.service: main process exited, code=exited, status=2/INVALIDARGUMENT
Sep  2 17:52:21 tvikgge systemd: Unit telegraf.service entered failed state.
Sep  2 17:52:21 tvikgge systemd: telegraf.service failed.
Sep  2 17:52:21 tvikgge systemd: telegraf.service holdoff time over, scheduling restart.
Sep  2 17:52:21 tvikgge systemd: Stopped Telegraf.
Sep  2 17:52:21 tvikgge systemd: Starting Telegraf...

Ideally, this race condition should be handled upstream in paho.mqtt.golang, rather than requiring workarounds at the Telegraf plugin level.

This issue is already reported in the Telegraf issue: influxdata/telegraf#17564 (reference if needed).

Thank you for your time and attention to this issue.

[MattBrittan]

Summarising the issue.

When options.AutoAckDisabled = true and a QOS1/2 PUBLISH is received it will not be acknowledged automaticallty. Instead the message passed to the user will include a function returned by ackFunc(ackInChan, client.persist, message):

// ackFunc acknowledges a packet
// WARNING the function returned must not be called if the comms routine is shutting down or not running
// (it needs outgoing comms in order to send the acknowledgement). Currently this is only called from
// matchAndDispatch which will be shutdown before the comms are
func ackFunc(oboundP chan *PacketAndToken, persist Store, packet *packets.PublishPacket) func() {
	return func() {
		switch packet.Qos {
		case 2:
			pr := packets.NewControlPacket(packets.Pubrec).(*packets.PubrecPacket)
			pr.MessageID = packet.MessageID
			DEBUG.Println(NET, "putting pubrec msg on obound")
			oboundP <- &PacketAndToken{p: pr, t: nil}
			DEBUG.Println(NET, "done putting pubrec msg on obound")
		case 1:
			pa := packets.NewControlPacket(packets.Puback).(*packets.PubackPacket)
			pa.MessageID = packet.MessageID
			DEBUG.Println(NET, "putting puback msg on obound")
			persistOutbound(persist, pa)
			oboundP <- &PacketAndToken{p: pa, t: nil}
			DEBUG.Println(NET, "done putting puback msg on obound")
		case 0:
			// do nothing, since there is no need to send an ack packet back
		}
	}
}

If the above function is called after the connection drops then ackInChan wil be closed leading to a panic.

I added the warning at the top back in 2019 and did some work in #505 to resolve an issue similar to this one (but without manual acknowledgments). Because the Ack() function for a message can be called at anytime, or not called at all, we cannot delay the shutdown/reconnection until all outstanding Ack() functions have been called. Unfortunately this gets complicated because there are a range of possibilities when Ack is called:

• The connection is still up (and ACK can/should be sent)
• The connection is down but session is still active (ACK should be queued if QOS2)
• The connection is down and session closed (ACK is ignored)
• New connection is up and session is still active (ACK should sent)
• New connection is up but its a different session (ACK is ignored)
• Connection is closed and client closed down (stores etc closed)

A further complication is ClientOptions.Order (messages should be acknowledged in the order they are received).

This means that resolving the issue properly will not be simple. I suspect that we might want an initial fix that just drops any ACK call after the connection has been lost, I'm thinking something along the lines of:

• Change ackFunc(oboundP chan *PacketAndToken, persist Store,... to ackFunc(sendAck func(p PacketAndToken),,...
• Write the sendAck function such that it only sends the Ack if the connection is still up

This is not ideal, but fixing the issue properly is likely to require significant changes. This is not something I currently have time to look at so I'll tag this issue as help wanted.

[lnxpgn]

Got it, thanks for the thorough breakdown. Appreciate it.

[webmogul1]

I am using Telegraf 1.36.1 and I having a similar(?) issue.
Whenever telegraf tries to connect to an IP that is up, but no mqqt is found, it stops and does not start again.
If IP is down, it does not stop.

[MattBrittan]

I was able to duplicate this issue in a test (Test_Ack_After_Disconnect) and believe it's addressed in #729 (part of release 1.5.1). I would appreciate it if you could test with this release.

Note: This is a workaround to prevent the panic; I don't necessarily see it as a "fix". Ideally the manual Ack should work regardless of the connection status (see my comments above). However I believe that would require some fairly major changes and is not something I have time to implement.

@webmogul1 unfortunately I don't use Telegraf so cannot really comment on whether you are seeing the same issue or not (but I would expect you to see a panic as opposed to just not reconnecting). If you wish to raise the issue here I would need you to do some debugging and submit logs etc (by that I mean the paho logs, or messages referencing this library, as opposed to generic Telegraf logs).

[lnxpgn]

Thank you for the response and for the update in release 1.5.1. I plan to test with release 1.5.1 next week by rebuilding Telegraf and will let you know the results after a few days of runtime.

[lnxpgn]

Just finished testing with paho.mqtt.golang v1.5.1 inside Telegraf 1.36.1. Here are the results:

1. Telegraf config

telegraf.d/input-mqtt-server-1.conf

[[inputs.mqtt_consumer]]
  servers = ["tcp://mqtt-server-1:9009"]
  topics = [
    "sensor/#",
  ]
  topic_tag = ""
  qos = 2
  keepalive = "30s"
  ping_timeout = "10s"
  persistent_session = true
  client_id = "t-mqtt-server-1-n10"
  username="xxx"
  password="xxx"
  data_format = "json"
  json_time_key = "time"
  client_trace = true

Added client_trace = true.

2. Telegraf code changes

--- telegraf-1.36.1/plugins/inputs/mqtt_consumer/mqtt_consumer.go       2025-09-09 08:16:32.000000000 +0800
+++ telegraf/plugins/inputs/mqtt_consumer/mqtt_consumer.go      2025-09-22 17:14:34.308018040 +0800
@@ -88,7 +88,7 @@
                mqtt.ERROR = log
                mqtt.CRITICAL = log
                mqtt.WARN = log
-               mqtt.DEBUG = log
+               // mqtt.DEBUG = log
        }

        if m.PersistentSession && m.ClientID == "" {

Commented this line due to excessive debug output.

3. paho.mqtt.golang library changes

--- paho.mqtt.golang-1.5.1/router.go    2025-09-16 12:09:00.000000000 +0800
+++ paho.mqtt.golang/router.go  2025-09-23 09:53:24.998032850 +0800
@@ -149,7 +150,7 @@
                if sendAckChan != nil {
                        sendAckChan <- ack
                } else {
-                       DEBUG.Println(ROU, "matchAndDispatch received acknowledgment after processing stopped (ACK dropped).")
+                       ERROR.Println(ROU, "matchAndDispatch received acknowledgment after processing stopped (ACK dropped).")
                }
        }

Changed this to error log due to excessive debug output.

4. Telegraf startup command

./telegraf -config myconf/telegraf/telegraf.conf -config-directory myconf/telegraf/telegraf.d --debug

5. Logs

No panic, the changes appear to be working as expected.

2025-09-23T02:19:49Z I! Loading config: myconf/telegraf/telegraf.conf
2025-09-23T02:19:49Z I! Loading config: myconf/telegraf/telegraf.d/input-sensor.conf
2025-09-23T02:19:49Z I! Loading config: myconf/telegraf/telegraf.d/output-stdout.conf
2025-09-23T02:19:49Z I! Loading config: myconf/telegraf/telegraf.d/processor-sensor.conf
2025-09-23T02:19:49Z I! Starting Telegraf 1.36.1- (customized) brought to you by InfluxData the makers of InfluxDB
2025-09-23T02:19:49Z I! Available plugins: 2 inputs, 0 aggregators, 2 processors, 5 parsers, 4 outputs, 0 secret-stores
2025-09-23T02:19:49Z I! Loaded inputs: mqtt_consumer
2025-09-23T02:19:49Z I! Loaded aggregators:
2025-09-23T02:19:49Z I! Loaded processors: starlark
2025-09-23T02:19:49Z I! Loaded secretstores:
2025-09-23T02:19:49Z I! Loaded outputs: file
2025-09-23T02:19:49Z I! Tags enabled:
2025-09-23T02:19:49Z I! [agent] Config: Interval:30s, Quiet:false, Hostname:"", Flush Interval:10s
2025-09-23T02:19:49Z W! [agent] The default value of 'skip_processors_after_aggregators' will change to 'true' with Telegraf v1.40.0! If you need the current default behavior, please explicitly set the option to 'false'!
2025-09-23T02:19:49Z D! [agent] Initializing plugins
2025-09-23T02:19:49Z D! [agent] Connecting outputs
2025-09-23T02:19:49Z D! [agent] Attempting connection to [outputs.file]
2025-09-23T02:19:49Z D! [agent] Successfully connected to outputs.file
2025-09-23T02:19:49Z D! [agent] Starting service inputs
2025-09-23T02:19:49Z I! [inputs.mqtt_consumer] Connected [tcp://mqtt-server-1:9009]
2025-09-23T02:19:49Z D! [inputs.mqtt_consumer] Session found [tcp://mqtt-server-1:9009]
2025-09-23T02:19:58Z D! [outputs.file] Wrote batch of 1000 metrics in 7.376686ms
2025-09-23T02:19:58Z D! [outputs.file] Buffer fullness: 0 / 10000 metrics
2025-09-23T02:19:59Z D! [outputs.file] Wrote batch of 613 metrics in 3.435133ms
2025-09-23T02:19:59Z D! [outputs.file] Buffer fullness: 0 / 10000 metrics
...
2025-09-23T02:23:19Z D! [inputs.mqtt_consumer] [client]  Connect comms goroutine - error triggeredEOF
2025-09-23T02:23:19Z D! [inputs.mqtt_consumer] [client]  BUG BUG BUG reconnection function is nil<nil>
2025-09-23T02:23:19Z D! [inputs.mqtt_consumer] [client]  status is already disconnected
2025-09-23T02:23:19Z E! [inputs.mqtt_consumer] Error in plugin: connection lost: EOF
2025-09-23T02:23:19Z D! [inputs.mqtt_consumer] Disconnected [tcp://mqtt-server-1:9009]
2025-09-23T02:23:19Z D! [outputs.file] Wrote batch of 180 metrics in 1.063349ms
2025-09-23T02:23:19Z D! [outputs.file] Buffer fullness: 0 / 10000 metrics
2025-09-23T02:23:29Z D! [outputs.file] Buffer fullness: 0 / 10000 metrics
2025-09-23T02:23:30Z D! [inputs.mqtt_consumer] Connecting [tcp://mqtt-server-1:9009]
2025-09-23T02:23:30Z I! [inputs.mqtt_consumer] Connected [tcp://mqtt-server-1:9009]
2025-09-23T02:23:30Z D! [inputs.mqtt_consumer] [store]   memorystore del: message1not found
2025-09-23T02:23:38Z D! [outputs.file] Wrote batch of 1000 metrics in 6.763895ms
2025-09-23T02:23:38Z D! [outputs.file] Buffer fullness: 0 / 10000 metrics
2025-09-23T02:23:39Z D! [outputs.file] Wrote batch of 162 metrics in 1.134494ms
...
2025-09-23T02:32:28Z D! [inputs.mqtt_consumer] [client]  Connect comms goroutine - error triggeredread tcp 2.3.4.5:47848->1.2.3.4:9009: read: connection reset by peer
2025-09-23T02:32:28Z D! [inputs.mqtt_consumer] [client]  BUG BUG BUG reconnection function is nil<nil>
2025-09-23T02:32:28Z D! [inputs.mqtt_consumer] [client]  status is already disconnected
2025-09-23T02:32:28Z E! [inputs.mqtt_consumer] Error in plugin: connection lost: read tcp 2.3.4.5:47848->1.2.3.4:9009: read: connection reset by peer
2025-09-23T02:32:28Z D! [inputs.mqtt_consumer] Disconnected [tcp://mqtt-server-1:9009]
2025-09-23T02:32:29Z D! [outputs.file] Wrote batch of 105 metrics in 684.017µs
2025-09-23T02:32:29Z D! [outputs.file] Buffer fullness: 0 / 10000 metrics
2025-09-23T02:32:29Z D! [inputs.mqtt_consumer] [router]  matchAndDispatch received acknowledgment after processing stopped (ACK dropped).
2025-09-23T02:32:29Z D! [inputs.mqtt_consumer] [router]  matchAndDispatch received acknowledgment after processing stopped (ACK dropped).
2025-09-23T02:32:29Z D! [inputs.mqtt_consumer] [router]  matchAndDispatch received acknowledgment after processing stopped (ACK dropped).
2025-09-23T02:32:29Z D! [inputs.mqtt_consumer] [router]  matchAndDispatch received acknowledgment after processing stopped (ACK dropped).
...
2025-09-23T02:32:30Z D! [inputs.mqtt_consumer] Connecting [tcp://mqtt-server-1:9009]
2025-09-23T02:32:30Z I! [inputs.mqtt_consumer] Connected [tcp://mqtt-server-1:9009]
2025-09-23T02:32:30Z D! [inputs.mqtt_consumer] [store]   memorystore del: message1not found
2025-09-23T02:32:39Z D! [outputs.file] Wrote batch of 1000 metrics in 7.498859ms
2025-09-23T02:32:39Z D! [outputs.file] Buffer fullness: 0 / 10000 metrics
2025-09-23T02:32:39Z D! [outputs.file] Wrote batch of 87 metrics in 594.048µs
2025-09-23T02:32:39Z D! [outputs.file] Buffer fullness: 0 / 10000 metrics
...
2025-09-23T02:41:18Z D! [inputs.mqtt_consumer] [client]  Connect comms goroutine - error triggeredEOF
2025-09-23T02:41:18Z D! [inputs.mqtt_consumer] [client]  BUG BUG BUG reconnection function is nil<nil>
2025-09-23T02:41:18Z D! [inputs.mqtt_consumer] [client]  status is already disconnected
2025-09-23T02:41:18Z E! [inputs.mqtt_consumer] Error in plugin: connection lost: EOF
2025-09-23T02:41:18Z D! [inputs.mqtt_consumer] Disconnected [tcp://mqtt-server-1:9009]
2025-09-23T02:41:20Z D! [outputs.file] Wrote batch of 117 metrics in 1.714292ms
2025-09-23T02:41:20Z D! [outputs.file] Buffer fullness: 0 / 10000 metrics
2025-09-23T02:41:20Z D! [inputs.mqtt_consumer] [router]  matchAndDispatch received acknowledgment after processing stopped (ACK dropped).
2025-09-23T02:41:20Z D! [inputs.mqtt_consumer] [router]  matchAndDispatch received acknowledgment after processing stopped (ACK dropped).
2025-09-23T02:41:20Z D! [inputs.mqtt_consumer] [router]  matchAndDispatch received acknowledgment after processing stopped (ACK dropped).
...
2025-09-23T02:41:30Z D! [inputs.mqtt_consumer] Connecting [tcp://mqtt-server-1:9009]
2025-09-23T02:41:30Z D! [outputs.file] Buffer fullness: 0 / 10000 metrics
2025-09-23T02:41:30Z I! [inputs.mqtt_consumer] Connected [tcp://mqtt-server-1:9009]
2025-09-23T02:41:30Z D! [inputs.mqtt_consumer] [store]   memorystore del: message1not found
2025-09-23T02:41:38Z D! [outputs.file] Wrote batch of 1000 metrics in 6.781623ms
2025-09-23T02:41:38Z D! [outputs.file] Buffer fullness: 0 / 10000 metrics
2025-09-23T02:41:40Z D! [outputs.file] Wrote batch of 173 metrics in 1.750623ms
2025-09-23T02:41:40Z D! [outputs.file] Buffer fullness: 0 / 10000 metrics
...
2025-09-23T03:09:18Z D! [inputs.mqtt_consumer] [client]  Connect comms goroutine - error triggered during send PubEOF
2025-09-23T03:09:18Z D! [inputs.mqtt_consumer] [client]  BUG BUG BUG reconnection function is nil<nil>
2025-09-23T03:09:18Z D! [inputs.mqtt_consumer] [client]  status is already disconnected
2025-09-23T03:09:18Z E! [inputs.mqtt_consumer] Error in plugin: connection lost: EOF
2025-09-23T03:09:18Z D! [inputs.mqtt_consumer] Disconnected [tcp://mqtt-server-1:9009]
2025-09-23T03:09:20Z D! [outputs.file] Wrote batch of 128 metrics in 882.528µs
2025-09-23T03:09:20Z D! [outputs.file] Buffer fullness: 0 / 10000 metrics
2025-09-23T03:09:20Z D! [inputs.mqtt_consumer] [router]  matchAndDispatch received acknowledgment after processing stopped (ACK dropped).
2025-09-23T03:09:20Z D! [inputs.mqtt_consumer] [router]  matchAndDispatch received acknowledgment after processing stopped (ACK dropped).
2025-09-23T03:09:20Z D! [inputs.mqtt_consumer] [router]  matchAndDispatch received acknowledgment after processing stopped (ACK dropped).
...
2025-09-23T03:50:20Z D! [inputs.mqtt_consumer] [pinger]  pingresp not received, disconnecting
2025-09-23T03:50:20Z D! [inputs.mqtt_consumer] [client]  BUG BUG BUG reconnection function is nil<nil>
2025-09-23T03:50:20Z D! [inputs.mqtt_consumer] [client]  status is already disconnected
2025-09-23T03:50:20Z E! [inputs.mqtt_consumer] Error in plugin: connection lost: pingresp not received, disconnecting
2025-09-23T03:50:20Z D! [inputs.mqtt_consumer] Disconnected [tcp://mqtt-server-1:9009]
2025-09-23T03:50:20Z D! [outputs.file] Wrote batch of 96 metrics in 638.169µs
2025-09-23T03:50:20Z D! [outputs.file] Buffer fullness: 0 / 10000 metrics
2025-09-23T03:50:30Z D! [inputs.mqtt_consumer] Connecting [tcp://mqtt-server-1:9009]
2025-09-23T03:50:30Z D! [outputs.file] Buffer fullness: 0 / 10000 metrics
2025-09-23T03:50:40Z D! [outputs.file] Buffer fullness: 0 / 10000 metrics
2025-09-23T03:50:45Z I! [inputs.mqtt_consumer] Connected [tcp://mqtt-server-1:9009]
2025-09-23T03:50:45Z D! [inputs.mqtt_consumer] [store]   memorystore del: message1not found
2025-09-23T03:50:50Z D! [outputs.file] Wrote batch of 644 metrics in 4.882519ms
2025-09-23T03:50:50Z D! [outputs.file] Buffer fullness: 1 / 10000 metrics
...
2025-09-23T05:51:19Z D! [agent] Stopping service inputs
2025-09-23T05:51:19Z D! [inputs.mqtt_consumer] Disconnecting [tcp://mqtt-server-1:9009]
2025-09-23T05:51:19Z D! [inputs.mqtt_consumer] Disconnected [tcp://mqtt-server-1:9009]
2025-09-23T05:51:19Z D! [agent] Input channel closed
2025-09-23T05:51:19Z D! [agent] Processor channel closed
2025-09-23T05:51:19Z I! [agent] Hang on, flushing any cached metrics before shutdown
2025-09-23T05:51:19Z D! [outputs.file] Wrote batch of 319 metrics in 1.793642ms
2025-09-23T05:51:19Z D! [outputs.file] Buffer fullness: 0 / 10000 metrics
2025-09-23T05:51:19Z I! [agent] Stopping running outputs
2025-09-23T05:51:19Z D! [agent] Stopped Successfully

[MattBrittan]

Excellent, thanks for testing. I'll close this issue (feel free to reopen if it reappears).