Slow down KafkaReceiver when downstream is busy (backpressure support)

[sokoide]

Component(s)

receiver/kafka

Is your feature request related to a problem? Please describe.

We use the OpenTelemetry Collector (otelcol) with the Kafka receiver as input and an OTLP exporter to send data into our trace backend.

When a spike of data arrives, the Kafka receiver can pull data very quickly into the pipeline.
However, if the trace backend is slow, the buffered data accumulates in otelcol memory, eventually leading to OOM.

Kafka --> Kafka Receiver --> (processors) --> OTLP Exporter --> Trace Backend

The memorylimiterprocessor can reject new data to avoid an OOM crash, but the Kafka receiver continues pulling from Kafka and pushing into the pipeline (retrying):

Kafka --> Kafka Receiver --> memorylimiterprocessor --> OTLP Exporter --> Trace Backend
                                             ^ rejection (error)

Related Issues:
open-telemetry/opentelemetry-collector#5456
#29410

Describe the solution you'd like

If the Kafka receiver receives a rejection (error) from any downstream component (processor or exporter), it should slow down fetching (consuming) from Kafka.

This introduces a natural form of backpressure: when the pipeline or backend is overloaded, the receiver reduces the pull rate.

Ideally, a specific error type (e.g. ErrBackendBusy) would be defined so that receivers can distinguish between "overloaded, slow down" and "fatal error, drop data".

Pseudo-code

err := nextConsumer.ConsumeTraces(ctx, traces)
if err != nil {
    if isMemoryLimitError(err) {
        r.logger.Warn("downstream busy, slowing down polling", zap.Error(err))
        time.Sleep(r.cfg.BackoffOnBusy) // default e.g. 100ms
        continue
    }
    r.logger.Error("consume failed", zap.Error(err))
}

func isMemoryLimitError(err error) bool {
    // TODO: ideally, use errors.Is(err, memorylimiter.ErrMemoryLimitExceeded)
    return strings.Contains(err.Error(), "memory limiter")
}

Describe alternatives you've considered

• Use the memorylimiterprocessor alone:
  This prevents OOM but does not propagate backpressure to Kafka, because the Kafka receiver keeps polling and resending rejected data.

• Increase batching or add queue processors:
  These can help absorb spikes but do not prevent OOM if the backend remains slow.

Additional context

Benefits:

• Improves stability under load spikes.
• Provides a natural feedback loop between downstream saturation and upstream ingestion.
• Lays the foundation for a standardized "busy error" type in otelcol that other receivers could also respect.

Tip

_{React with 👍 to help prioritize this issue. Please use comments to provide useful context, avoiding +1 or me too, to help us triage it. Learn more here.}

[github-actions]

Pinging code owners:

• receiver/kafka: @pavolloffay @MovieStoreGuy @axw

See Adding Labels via Comments if you do not have permissions to add labels yourself.

[axw]

When a spike of data arrives, the Kafka receiver can pull data very quickly into the pipeline.
However, if the trace backend is slow, the buffered data accumulates in otelcol memory, eventually leading to OOM.

Can you please elaborate on this? Kafka records are processed synchronously, so backpressure should naturally occur. However, this is per-partition. Is the issue here that you have many topics/partitions?

I'm not sure that we can simply slow down consuming for all partitions because one of them is slow. They may be routed to different pipelines/backends.

[sokoide]

@axw, thank you for taking a look. We often have a huge data (spike for a while) in Kafka across topic-partitions (toppars). Since Kafka receiver is very fast, it pulls many of them into OTELCOL which takes more time to export.
Because we don't know what toppar is causing it, we wonder if we can slow down entire Kafka consumer (OTEL KafkaReceiver) when the OTEL downstream (memorylimitterprocessor in the case) is busy.

different pipelines/backends
I agree, we shouldn't simply slow it down because it can have multiple pipelines. Does it make sense if we add a flag to slow it down or do nothing (current implementation)? That'd help our use case - our OTELCOL for this is dedicated to pull from Kafka into the Tracing backend.

[axw]

@sokoide I'm not sure, because I don't know how the solution will look yet.

We already have some code that does retry with backoff here:

opentelemetry-collector-contrib/receiver/kafkareceiver/consumer_franz.go

Lines 484 to 497 in a6f867b

	if pc.backOff != nil && !consumererror.IsPermanent(err) {
	backOffDelay := pc.backOff.NextBackOff()
	if backOffDelay != backoff.Stop {
	pc.logger.Info("Backing off due to error from the next consumer.",
	zap.Error(err),
	zap.Duration("delay", backOffDelay),
	)
	select {
	case <-pc.ctx.Done():
	return context.Cause(pc.ctx)
	case <-time.After(backOffDelay):
	continue
	}
	}

I'm having trouble understanding why memorylimiterprocessor + that retry-backoff isn't good enough.

[sokoide]

@axw, thank you for sharing! I wasn't aware of it. I found the commit 0dc57b64fbcab306a9fe663daf3e4b27a5aed3fe from your keyword. My quick test went well with "Backing off due to error from the next consumer" when the backend is busy. It looked it stopped receiving from Kafka until "Memory usage back within limits. Resuming normal operation" from the memorylimiter.

[axw]

@sokoide great! Thanks for reporting back. If you have any suggestions for improvements to this feature, please do open another issue.