Two changes to the OpenTelemetry Kafka receiver that stop one slow tenant stalling the rest

Elastic speaks OpenTelemetry natively. Send traces, logs, and metrics over OTLP straight into Elasticsearch, no proprietary agents required. See how it fits together, try it for free in the cloud, or run it locally.

We contributed two changes to the OpenTelemetry Kafka receiver so that a slow tenant only delays itself. Each Kafka partition now gets its own worker, and a new processor caps how many sending-queue slots any single tenant can hold. Together that is enough for multi-tenant isolation on one shared receiver. When a tenant's Elasticsearch cluster is under load, its records wait on its own partition while the rest keep fetching and delivering. Nothing is dropped, and delivery stays at-least-once.

Elastic Cloud Managed Inputs runs several collectors that work together. In one of them, the Kafka receiver reads telemetry records from Kafka topics across tenants and hands them to the Elasticsearch exporter. Those topics hold the logs, metrics and traces that tenants send to Managed Inputs. A tenant can use more than one topic, but every topic belongs to a single tenant, and the exporter delivers each tenant's records to that tenant's own Elasticsearch cluster.

The diagram below follows one tenant through that path. Tenant A's logs and metrics topics feed a single OpenTelemetry Collector, the Kafka receiver reads the records, and the Elasticsearch exporter writes them to Tenant A's Elasticsearch cluster. Every tenant on the pod follows the same path through the same receiver, and sharing it is what turns one slow tenant into a noisy neighbor.

Why a shared Kafka consumer group creates noisy neighbors

Our Kafka receiver configuration relies on a single consumer group matching topics via a wildcard pattern. This surfaced an issue: telemetry streams from different tenants, such as managed_otlp.logs.a for Tenant A and managed_otlp.logs.b for Tenant B, were ingested and handled by the exact same consumer instance.

That consumer fetched records for every topic assigned to it, then processed the fetch one record at a time. It unmarshaled a record, handed it to the pipeline, and only then moved on to the next one. Because the fetch mixed tenants, Tenant A and Tenant B were handled sequentially in the same loop.

Each consumed record landed on the Elasticsearch exporter's sending queue, which is shared by every tenant on the pod. A worker pool drained this queue to send data to each tenant's Elasticsearch cluster concurrently.

However, a slow Elasticsearch cluster would crowd out the healthy ones. Workers took longer to finish sends for the lagging tenant, causing those records to pile up while the receiver kept adding more. Eventually, the shared queue filled up with the slow tenant's traffic, leaving no space for anyone else. Tenant A's records waited even when Tenant A's Elasticsearch was ready to take them.

With the queue full, the consumer loop inside the Kafka receiver blocked. It could not start the next record or fetch again, so healthy tenants on that consumer stopped making progress.

This illustration of the shared queue shows how the blocking happens. A single Kafka receiver loop reads records from both tenants and feeds them into one sending queue, so a backlog from one tenant blocks the loop for everyone.

Partition workers in the OpenTelemetry Kafka receiver

Our solution to fix this issue was to introduce independent partition processing to the Kafka receiver by giving each partition assigned to the consumer its own worker.

Instead of processing mixed tenant records in a single sequential loop, each worker now fetches from its own partition, unmarshals records one at a time, and passes them to the pipeline. As a result, if Tenant B's Elasticsearch cluster slows down, only Tenant B's partition worker stalls. Tenant A's worker continues fetching and processing uninterrupted.

The partition workers figure below shows the new layout: each tenant's partition has its own Kafka receiver worker. The workers no longer block each other, but both still push records into the same shared sending queue.

However, this change alone didn't fully resolve the issue. Because those independent workers still push records into the same shared sending queue, a slow tenant can still eventually fill that queue to capacity. The difference is in the loop itself. Healthy tenants are no longer stuck behind slow ones inside the consumer, so they keep making progress until the queue reaches capacity.

Capping per-tenant slots in the shared sending queue

To address the remaining queue contention between tenants, we introduced a processor positioned between the receiver and the exporter. This processor enforces an adaptive per-tenant limit on the number of slots a tenant can occupy in the shared sending queue.

Instead of a static allocation, each tenant holds a dynamic share of the exporter worker pool's in-flight slots, up to a configured ceiling. The processor adjusts that share as follows:

• A tenant starts with a minimal cap.
• The cap scales up incrementally with each successful request.
• On slow responses or retryable errors, the processor immediately halves the cap.
• Records are never dropped. When a tenant hits its active cap, subsequent flushes wait for an open slot, which applies backpressure to that tenant's partition worker and pauses fetches for that partition only.

As the tenant limiter diagram below shows, the limiter stands between the partition workers and the shared sending queue. It caps how many queue slots the slow tenant can fill, so the remaining slots stay available to healthy tenants.

Because the processor caps each tenant to a dynamic fraction of the queue's total capacity, a single degraded tenant can no longer consume all available buffers. If Tenant B's Elasticsearch cluster slows down and its queue usage reaches its active cap, the remaining queue capacity stays reserved for healthy tenants. Tenant A can continue fetching, processing, and queuing its records without interruption while Tenant B waits at the cap.

Tuning partition concurrency with max_in_flight

With tenants isolated from each other, we also improved throughput within each partition: the partition_processing.max_in_flight.records setting lets a worker unmarshal and pass several records to the pipeline at once.

The last image uses max_in_flight.records set to 2 as its example: one partition worker unmarshals two records at the same time and hands both to the pipeline, instead of finishing one before starting the next.

Bounding memory with max_in_flight.bytes

Record count alone doesn't bound memory. Setting only records: 64 can mean 64 small records or 2 GB of large ones, which is why we also added partition_processing.max_in_flight.bytes. It caps the data held in flight, so memory can be sized from configuration.

Offset marking and at-least-once delivery

Concurrency also changes what happens after a failure. Offsets are marked only along the contiguous prefix of records in fetch order, so records that finished but weren't marked are replayed after a crash or reassignment. Delivery is at-least-once: duplicates are possible, but records aren't lost.

Each change addresses a different part of the problem: