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

### 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:

```text
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):

```text
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:

```text
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/sarama@v1.60.0/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()
: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:

```text
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:

```text
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:

1. Run Telegraf with `outputs.kafka` against a multi-broker cluster.
2. Have the cluster go through broker restarts / leadership elections while
Telegraf is writing.
3. 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.