diff --git a/plugins/outputs/mqtt/mqtt.go b/plugins/outputs/mqtt/mqtt.go index 3bcf1b7d8..05978aa60 100644 --- a/plugins/outputs/mqtt/mqtt.go +++ b/plugins/outputs/mqtt/mqtt.go @@ -4,6 +4,7 @@ import ( "fmt" "strings" "sync" + "time" "github.com/influxdata/telegraf" "github.com/influxdata/telegraf/internal" @@ -25,6 +26,9 @@ var sampleConfig = ` # username = "telegraf" # password = "metricsmetricsmetricsmetrics" + ## Timeout for write operations. default: 5s + # timeout = "5s" + ## client ID, if not set a random ID is generated # client_id = "" @@ -149,7 +153,7 @@ func (m *MQTT) Write(metrics []telegraf.Metric) error { func (m *MQTT) publish(topic string, body []byte) error { token := m.client.Publish(topic, byte(m.QoS), false, body) - token.Wait() + token.WaitTimeout(m.Timeout.Duration) if token.Error() != nil { return token.Error() } @@ -159,6 +163,11 @@ func (m *MQTT) publish(topic string, body []byte) error { func (m *MQTT) createOpts() (*paho.ClientOptions, error) { opts := paho.NewClientOptions() + if m.Timeout.Duration < time.Second { + m.Timeout.Duration = 5 * time.Second + } + opts.WriteTimeout = m.Timeout.Duration + if m.ClientID != "" { opts.SetClientID(m.ClientID) } else {