influxdata / influxdata/telegraf

Telegraf not writing to Kafka if at least one message of the batch is intended for non-existing topic

Open
#10,864 2 comments 0 reactions 0 assignees View on GitHub
area/kafka bug
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.