influxdata / influxdata/telegraf
Telegraf not writing to Kafka if at least one message of the batch is intended for non-existing topic
- Dominant language
- Go
- Stars
- 17.8k
- Forks
- 5.8k
- Avg merge
- 1d 20h
- Merged PRs (30d)
- 161
Description
### Relevant telegraf.conf
```toml
[[outputs.kafka]]
brokers = ["localhost:9092"]
topic_tag = "X-Topic-Name"
exclude_topic_tag = true
```
### Logs from Telegraf
2022-03-21T16:13:03Z D! [outputs.kafka] Wrote batch of 1 metrics in 17.505578ms
2022-03-21T16:13:03Z D! [outputs.kafka] Buffer fullness: 0 / 1000000 metrics
2022-03-21T16:13:33Z D! [outputs.kafka] Buffer fullness: 0 / 1000000 metrics
2022-03-21T16:14:03Z D! [outputs.kafka] Buffer fullness: 0 / 1000000 metrics
2022-03-21T16:14:33Z D! [outputs.kafka] Wrote batch of 1 metrics in 1.716951ms
2022-03-21T16:14:33Z D! [outputs.kafka] Buffer fullness: 0 / 1000000 metrics
2022-03-21T16:15:03Z D! [outputs.kafka] Buffer fullness: 0 / 1000000 metrics
2022-03-21T16:10:36Z D! [outputs.kafka] Buffer fullness: 2 / 1000000 metrics
2022-03-21T16:10:36Z E! [agent] Error writing to outputs.kafka: kafka: Failed to produce message to topic test_topic_not_existing: kafka server: Request was for a topic or partition that does not exist on this broker.
2022-03-21T16:11:05Z D! [sarama] client/metadata fetching metadata for [test_topic_not_existing] from broker kafka:9093
2022-03-21T16:11:05Z D! [sarama] client/metadata found some partitions to be leaderless
2022-03-21T16:11:05Z D! [sarama] client/metadata retrying after 250ms... (3 attempts remaining)
2022-03-21T16:11:06Z D! [sarama] client/metadata fetching metadata for [test_topic_not_existing] from broker kafka:9093
2022-03-21T16:11:06Z D! [sarama] client/metadata found some partitions to be leaderless
2022-03-21T16:11:06Z D! [sarama] client/metadata retrying after 250ms... (2 attempts remaining)
2022-03-21T16:11:06Z D! [sarama] client/metadata fetching metadata for [test_topic_not_existing] from broker kafka:9093
2022-03-21T16:11:06Z D! [sarama] client/metadata found some partitions to be leaderless
2022-03-21T16:11:06Z D! [sarama] client/metadata retrying after 250ms... (1 attempts remaining)
2022-03-21T16:11:06Z D! [sarama] client/metadata fetching metadata for [test_topic_not_existing] from broker kafka:9093
2022-03-21T16:11:06Z D! [sarama] client/metadata found some partitions to be leaderless
2022-03-21T16:11:06Z D! [outputs.kafka] Buffer fullness: 2 / 1000000 metrics
2022-03-21T16:11:06Z E! [agent] Error writing to outputs.kafka: kafka: Failed to produce message to topic test_topic_not_existing: kafka server: Request was for a topic or partition that does not exist on this broker.
2022-03-21T16:11:35Z D! [sarama] client/metadata fetching metadata for [test_topic_not_existing] from broker kafka:9093
2022-03-21T16:11:35Z D! [sarama] client/metadata found some partitions to be leaderless
2022-03-21T16:11:35Z D! [sarama] client/metadata retrying after 250ms... (3 attempts remaining)
2022-03-21T16:11:36Z D! [sarama] client/metadata fetching metadata for [test_topic_not_existing] from broker kafka:9093
2022-03-21T16:11:36Z D! [sarama] client/metadata found some partitions to be leaderless
2022-03-21T16:11:36Z D! [sarama] client/metadata retrying after 250ms... (2 attempts remaining)
2022-03-21T16:11:36Z D! [sarama] client/metadata fetching metadata for [test_topic_not_existing] from broker kafka:9093
2022-03-21T16:11:36Z D! [sarama] client/metadata found some partitions to be leaderless
2022-03-21T16:11:36Z D! [sarama] client/metadata retrying after 250ms... (1 attempts remaining)
2022-03-21T16:11:36Z D! [sarama] client/metadata fetching metadata for [test_topic_not_existing] from broker kafka:9093
2022-03-21T16:11:36Z D! [sarama] client/metadata found some partitions to be leaderless
2022-03-21T16:11:36Z D! [outputs.kafka] Buffer fullness: 2 / 1000000 metrics
### System info
1.20.3
### Docker
_No response_
### Steps to reproduce
1. Send a metric intended for non-existing topic
2.
3.
...
### Expected behavior
Telegraf should discard the "faulty" metrics or be able to flush only "genuine" metrics from the batch/buffer.
### Actual behavior
If at least one message from the batch is intended for a non-existing topic, Telegraf and Sarama will complain about it but leave the buffer untouched. After encountering such metric(s), the buffer will never be flushed anymore and the tool needs to be restarted.
### Additional info
_No response_
Contributor guide
Research direction
Start by tracing the outputs.kafka batch-flushing path and how Sarama reports a failed produce for a missing topic. Reproduce the mixed-topic batch described in the issue, then verify that the faulty metric no longer blocks genuine metrics from being flushed and that the relevant error behavior is covered by tests.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- go, kafka
- Domain
- data-engineering, stream-processing
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 35/100