kafka_consumer: one unresponsive output stalls consumption for ALL outputs (same root cause as #16537, but for Kafka)

[dburton-influxdata]

Summary

inputs.kafka_consumer stops fetching/committing new messages entirely once max_undelivered_messages in-flight metrics are waiting on any single output, even if every other configured output is healthy and keeping up. In practice this means one misconfigured, slow, or dead output can silently stall an entire Kafka consumer pipeline - including all the outputs that were working fine.

This is the exact same architectural issue as #16537 ("MQTT input sent to two different outputs completely stops reporting if one output unresponsive"), just reproduced against kafka_consumer instead of mqtt_consumer. I'm filing this as a separate, Kafka-specific issue because:

1. kafka_consumer is one of the most widely used tracking inputs in Telegraf, and this failure mode is a real production incident pattern for Kafka-based pipelines specifically (root-caused a customer support case).
2. MQTT input sent to two different outputs completely stops reporting if one output unresponsive #16537 has been open since Feb 2025 with a concrete fix proposed (min_delivery_count) that appears to have stalled - no linked PR, no update since March 2025.
3. Framing this against Kafka specifically (with its offset-commit semantics) suggests an additional design option beyond what's discussed in MQTT input sent to two different outputs completely stops reporting if one output unresponsive #16537, described below.

I'd like to request this be treated as a real reliability gap (a "bug" from an operator's perspective, even though it's intentional by design) and prioritized as a fix, rather than left as an accepted workaround. Given how widely kafka_consumer is deployed, the current behavior surprises operators and there is no way to avoid it short of running duplicate Telegraf processes.

Root cause (confirmed by reading the source)

• metric.WithGroupTracking (metric/tracking.go) attaches a single shared trackingData struct with one reference counter (Rc) to every metric fanned out to N outputs. Accept()/Reject()/Drop() on any output's copy of the metric just decrement this one shared counter - there's no per-output distinction.
• The tracking metric only fires its notifyFunc (delivery notification) when Rc reaches zero, i.e. every output has accepted/rejected/dropped its copy.
• plugins/inputs/kafka_consumer/kafka_consumer.go's consumerGroupHandler.onDelivery listens on acc.Delivered() and only then permits the semaphore slot (bounded by max_undelivered_messages, default 1000) to free up and the corresponding Kafka offset to be committed.
• Net effect: if Output A is dead, its copies of tracked metrics never get Accept()/Reject()'d, Rc never reaches 0, onDelivery never fires for those metrics, the undelivered-message semaphore fills up, and kafka_consumer stops calling ConsumeClaim's message-processing loop for all partitions - regardless of the fact that Output B is healthy and would happily keep accepting messages.

Reproduction (real Kafka broker, not a hypothetical)

Built and ran this against Apache Kafka 4.3.1 (KRaft mode) + Telegraf 1.40.0:

[[inputs.kafka_consumer]]
  brokers = ["localhost:9094"]
  topics = ["test-multioutput"]
  consumer_group = "telegraf-test-c"
  max_undelivered_messages = 5   # lowered from the 1000 default purely to reproduce quickly

[[outputs.influxdb_v3]]         # HEALTHY - real, reachable endpoint
  urls = ["http://localhost:8181"]
  database = "telegraf_multi_output_test"

[[outputs.influxdb_v3]]         # BROKEN - unreachable, simulates a dead endpoint
  urls = ["http://localhost:19999"]
  database = "telegraf_multi_output_test"

Published 30 line-protocol messages to the topic. Result:

D! [outputs.influxdb_v3] Buffer fullness: 5 / 200 metrics    <- BROKEN output, capped at exactly max_undelivered_messages
E! [outputs.influxdb_v3] Writing to "http://localhost:19999" failed: ... connection refused
D! [outputs.influxdb_v3] Buffer fullness: 0 / 200 metrics    <- HEALTHY output, receives NOTHING further

Only the first 5 of 30 published messages ever reached the fully-healthy, fully-reachable output. The remaining 25 were never even fetched from Kafka - kafka_consumer paused entirely once the broken output's undelivered count hit the limit.

For contrast, the same setup with only polling inputs (cpu, mem, disk, docker - i.e. anything using the plain, untracked Accumulator.AddFields() path instead of WithTracking/WithGroupTracking) shows zero cross-output effect: a dead output just accumulates its own undelivered buffer independently while every healthy output keeps writing every flush interval with no delay. So this is specific to tracking inputs (kafka_consumer, mqtt_consumer, nats_consumer, amqp_consumer, etc.), not a general Telegraf output-handling issue.

Current workaround (and why it's not a good long-term answer)

Run one Telegraf instance per output, each with its own kafka_consumer and its own consumer_group, so Kafka's independent per-consumer-group offset tracking gives full isolation. This works, but:

• Doubles (or N×'s) Kafka consumer load on the broker for every additional output.
• Doubles the number of Telegraf processes to deploy, configure, and monitor.
• Is a workaround we now have to proactively recommend to every customer running multiple outputs off a Kafka source - which suggests the framework should handle this instead of pushing operational complexity onto every user.

Proposed enhancement(s)

Two complementary options were discussed on #16537; I'd suggest implementing at least one, ideally with both available since they solve slightly different needs:

Option 1 - min_delivery_count on the tracking input (as proposed in #16537's discussion, apparently never implemented):

[[inputs.kafka_consumer]]
  # ...
  min_delivery_count = 1   # commit/ack once N outputs have accepted, instead of waiting for all
  # default: -1 (sentinel meaning "wait for all outputs", i.e. current behavior - fully backward compatible)

Option 2 - per-output critical/required_for_delivery flag (finer-grained, and arguably a better fit for the common "primary + best-effort secondary" pattern our customer and #16537's reporter both hit):

[[outputs.influxdb_v3]]
  urls = ["http://primary:8181"]
  # implicitly required for delivery tracking (default true, current behavior)

[[outputs.influxdb_v3]]
  urls = ["http://secondary-analytics:8181"]
  required_for_delivery = false   # a slow/dead copy of this output should never block input consumption

This maps more directly onto the real-world scenario (one output is genuinely load-bearing/critical, another is a "nice to have" secondary/analytics copy) than a bare count, and doesn't require the user to reason about how many outputs must ack - just which ones matter for backpressure purposes.

Either approach should default to the current wait-for-all behavior for full backward compatibility.

Related

• MQTT input sent to two different outputs completely stops reporting if one output unresponsive #16537 - same root cause, reproduced via mqtt_consumer, open since Feb 2025, proposed fix stalled with no linked PR since March 2025
• metric/tracking.go, models/running_output.go, plugins/inputs/kafka_consumer/kafka_consumer.go (onDelivery, MaxUndeliveredMessages)

Happy to help test/validate against a real Kafka broker + multi-output setup if useful - I have a reproducible harness (Docker Kafka broker in KRaft mode, healthy/broken InfluxDB 3 output pair) already built for this investigation.

[srebhan]

@dburton-influxdata changing the tracking metrics is an architectural change in Telegraf and needs some thoughts on possible consequences as it touches the accumulator code, one of the core moving parts of Telegraf. What I'm trying to say is this will likely not happen anytime soon as we first need a spec before we start implementing things.