#19446·telegraf

outputs.kafka can block forever in SendMessages, wedging the flush loop and preventing graceful shutdown

Author: FireBurnCreated Aug 12, 2026Updated Sep 18, 2026

Relevant telegraf.conf

toml

[agent]
  interval = "30s"
  round_interval = true
  metric_batch_size = 1000
  metric_buffer_limit = 10000
  collection_jitter = "0s"
  flush_interval = "20s"
  flush_jitter = "0s"
  logfile = "/var/log/telegraf/telegraf.log"
  logfile_rotation_max_size = "10MB"
  logfile_rotation_max_archives = 3

[[inputs.cpu]]
  percpu = false
  totalcpu = true

[[inputs.diskio]]
[[inputs.mem]]
[[inputs.system]]
[[inputs.linux_sysctl_fs]]
[[inputs.disk]]

[[outputs.kafka]]
  brokers = ["kafka-a.example.com:9092", "kafka-b.example.com:9092"]
  topic = "metrics"
  data_format = "influx"

(hostnames/topic sanitised)

Logs from Telegraf

The Kafka cluster goes through occasional broker maintenance and partition leadership elections; errors like this show up in the Telegraf log in those windows and are normally recovered from:

2026-07-20T14:09:25Z E! [agent] Error writing to outputs.kafka: kafka: Failed to produce message to topic metrics: kafka server: In the middle of a leadership election, there is currently no leader for this partition and hence it is unavailable for writes

On one instance the output never recovered. From that point on the log is nothing but this, every flush interval, for weeks (6800+ occurrences in the retained log):

2026-08-12T09:26:20Z W! [agent] ["outputs.kafka"] did not complete within its flush interval
2026-08-12T09:26:40Z W! [agent] ["outputs.kafka"] did not complete within its flush interval

I sent the process SIGQUIT to capture a goroutine dump. The relevant goroutines (full dump available on request; build paths sanitised):

The write goroutine had been sitting inside sarama.SyncProducer.SendMessages() for 21475 minutes (~15 days) waiting on a channel receive:

goroutine 779378 gp=0xe392ca112c0 m=nil [chan receive, 21475 minutes]:
runtime.gopark(0xe392bffccb0?, 0x5000000002000?, 0x80?, 0x1a?, 0x7ffbda08fa00?)
	/usr/local/go/src/runtime/proc.go:462 +0xce
runtime.chanrecv(0xe392ca8d0a0, 0xe392b37dbf0, 0x1)
	/usr/local/go/src/runtime/chan.go:667 +0x4ae
runtime.chanrecv1(0x5000000000129?, 0x90958a0?)
	/usr/local/go/src/runtime/chan.go:509 +0x12
github.com/IBM/sarama.(*syncProducer).SendMessages(0xe392bc4d968, {0xe392bc4a008, 0x3e8, 0x3e8})
	/go/pkg/mod/github.com/!i!b!m/[email protected]/sync_producer.go:157 +0x165
github.com/influxdata/telegraf/plugins/outputs/kafka.(*Kafka).Write(0xe392b07b008, {0xe392bc60008, 0x3e8, 0x0?})
	/build/telegraf/plugins/outputs/kafka/kafka.go:199 +0x35d
github.com/influxdata/telegraf/models.(*RunningOutput).writeMetrics(0xe392a8c7680, {0xe392bc60008, 0x3e8, 0x3e8})
	/build/telegraf/models/running_output.go:397 +0x114
github.com/influxdata/telegraf/models.(*RunningOutput).doTransaction(0xe392a8c7680)
	/build/telegraf/models/running_output.go:377 +0x55
github.com/influxdata/telegraf/models.(*RunningOutput).Write(0xe392a8c7680)
	/build/telegraf/models/running_output.go:341 +0x3c6
github.com/influxdata/telegraf/models.(*RunningOutput).Write-fm()
	<autogenerated>:1 +0x17
github.com/influxdata/telegraf/agent.(*Agent).flushOnce.func1()
	/build/telegraf/agent/agent.go:922 +0x23
created by github.com/influxdata/telegraf/agent.(*Agent).flushOnce in goroutine 8
	/build/telegraf/agent/agent.go:921 +0x9e

The output's flush loop is wedged inside flushOnce waiting for that write to return — flushOnce only returns when the write function returns, so it can never observe shutdown either:

goroutine 8 gp=0xe392a590000 m=nil [select]:
runtime.gopark(0xe392c97bcb0?, 0x2?, 0x0?, 0x0?, 0xe392c97bc84?)
	/usr/local/go/src/runtime/proc.go:462 +0xce
runtime.selectgo(0xe392c97bcb0, 0xe392c97bc80, 0x2?, 0x0, 0x1?, 0x1)
	/usr/local/go/src/runtime/select.go:351 +0xaa5
github.com/influxdata/telegraf/agent.(*Agent).flushOnce(0xe392c97be48?, 0xe392a8c7680, 0xe392abf4230, 0xe392badc790)
	/build/telegraf/agent/agent.go:926 +0x1a6
github.com/influxdata/telegraf/agent.(*Agent).flushLoop(0xe392bb90390, {0xb0ea768, 0xe392abf4190}, 0xe392a8c7680, 0xe392abf4230)
	/build/telegraf/agent/agent.go:909 +0x212
github.com/influxdata/telegraf/agent.(*Agent).runOutputs.func1(0xe392a8c7680)
	/build/telegraf/agent/agent.go:859 +0xb7
created by github.com/influxdata/telegraf/agent.(*Agent).runOutputs in goroutine 31
	/build/telegraf/agent/agent.go:853 +0xbc

And the main goroutine is in Agent.Run waiting on the worker WaitGroup (28848 minutes = process uptime, ~20 days), which is why a graceful shutdown can never complete while the write is stuck:

goroutine 1 gp=0xe392a5701e0 m=nil [sync.WaitGroup.Wait, 28848 minutes]:
sync.(*WaitGroup).Wait(0xe392b498550)
	/usr/local/go/src/sync/waitgroup.go:206 +0x85
github.com/influxdata/telegraf/agent.(*Agent).Run(0xe392bb90390, {0xb0ea768, 0xe392b411860})
	/build/telegraf/agent/agent.go:209 +0xb06
main.(*Telegraf).runAgent(0xe392b194b00, {0xb0ea768, 0xe392b411860}, 0x0?)
	/build/telegraf/cmd/telegraf/telegraf.go:572 +0x1885
main.(*Telegraf).reloadLoop(0xe392b194b00)
	/build/telegraf/cmd/telegraf/telegraf.go:208 +0x26b

System info

Telegraf 1.39.2 (built from source), Linux x86_64, github.com/IBM/sarama v1.60.0

Docker

n/a

Steps to reproduce

I don't have a deterministic reproducer — this hits rarely, across a fleet of agents, typically around Kafka broker maintenance/partition leadership churn:

Run Telegraf with outputs.kafka against a multi-broker cluster.
Have the cluster go through broker restarts / leadership elections while Telegraf is writing.
Very occasionally, a batch handed to SyncProducer.SendMessages() never receives a terminal success/error result, and the call never returns.

The stuck state itself is easy to simulate in code by blocking SendMessages() (that's what the regression tests in the linked PR do).

Expected behavior

A single stuck write should not permanently wedge the output. Telegraf should be able to time the write out (or at least abandon it), resume flushing buffered metrics, and remain able to shut down gracefully.

Actual behavior

SendMessages() blocked for ~15 days. The flush loop was wedged the entire time (did not complete within its flush interval every 20s), the metric buffer filled and dropped everything beyond metric_buffer_limit, and the agent could not be shut down gracefully — flushOnce only returns when the write returns, and Agent.Run waits on the flush worker. The process had to be killed.

Additional info

sarama.SyncProducer.SendMessages() takes no context and waits for every message to receive a terminal result from Sarama's async producer machinery. Sarama's dial/read/write network timeouts were left at their defaults, but they don't guarantee a terminal result for every message in every failure mode — and whatever the broker-side trigger was here, the result was a receive that never completed. Since Telegraf's Output.Write() interface has no context either, there is no cancellation path from the agent down to the producer.

#11427 and #8349 (both closed) describe what looks like the same failure mode: outputs.kafka wedged after a broker event, never recovering without a restart.

I have a patch that adds an optional context-aware output interface plus a per-output write_timeout option, and implements it for outputs.kafka — PR to follow shortly.

Source: influxdata/telegraf

View original on GitHub View discussion on GitHub