In this post we’re going to see how share.acquire.mode=record_limit combined with:

• fewer consumers than partitions
• and various cases of “partition skew”

…can result in subpar performance with share groups.

I stumbled on these issues when running large sets of dimensional tests with Dimster’s explore-limits mode, which finds the highest sustainable throughput while staying within a target end-to-end latency target. There was a specific subset of the tests that explore-limits mode would consistently fail to complete, and they all happened to be with record_limit and a consumer count lower than the partition count. In this test, we’ll understand why Dimster had such a hard time with this combination.

Some background on share group internals

Kafka share groups have two methods of acquiring records:

• share.acquire.mode=batch_optimized
• share.acquire.mode=record_limit

I already explained the difference in Kafka Share Groups and Parallelizing Consumption - Part 2: Producer Batches and share.acquire.mode but let’s just cover it again.

Share consumers are assigned partitions as part of the share group protocol. It works similarly to the consumer group protocol, except that multiple consumers can be assigned to the same partition.

With batch_optimized, share consumers acquire records in whole batches, using max.poll.records as a soft cap. Furthermore, a share consumer assigned multiple partitions across multiple brokers will send fetch requests to each of those brokers, concurrently.

With record_limit, share consumers acquire records as slices of batches, where the size of the slice is determined by max.poll.records (now a strict cap). If you set max.poll.records=10 but the relevant batch contains 32, then only a slice of 10 records is acquired (though the whole batch is transmitted over the wire). Furthermore, a share consumer assigned multiple partitions across multiple brokers will send fetch requests round-robin (one-at-a-time) across those brokers. Each time you call poll, it will fetch from the next broker.

Dimster’s failing tests

Dimster consistently did not complete explore-limits tests with record_limit and fewer consumers than partitions. The issue is that during various phases of an explore-limits test, lag can build very quickly if producers shoot past the capacity of the consumers. Dimster sees this and attempts to drain the lag before it resumes with a lower producer rate.

The drain works by pausing the producers, temporarily removing any consumer processing time (if configured) and then resuming with a lower producer rate. However, with record_limit and fewer consumers than partitions, this lag drain would basically stall as the consumption rate would end up just a trickle (such that it would take hours to drain the backlog that had accumulated).

So I ran some backlog drain tests to understand what was going on and discovered what I’ll refer to as pathological fetch waits.

Pathological fetch waits

Imagine one share consumer and a topic of 10 partitions spread across 3 brokers. Imagine if all the producers sent records to only one partition, leaving the other 9 consistently empty. What sub-optimal share consumer behavior might we see?

Let’s go through it. Remember, with record_limit, fetches to brokers are round-robin if a consumer is assigned multiple partitions (on different brokers):

1. Consumer sends a fetch to broker-0 (which hosts partitions 0, 3, 6, 9) and gets back some records for partition 0.
2. Poll is called again, triggering a fetch to broker-1 (which hosts partitions 1, 4, 7), but there are no records.
3. Poll is called again, triggering a fetch to broker-2 (which hosts partitions 3, 5, 8) but there are no records.
4. Poll is called again, triggering a fetch to broker-0, returning more records of partition 0.

So what’s the problem? Can you see it yet?

The problem is fetch.max.wait.ms. It defaults to 500. Yes that’s right, steps 2 and 3 took 1 second to complete and returned no records! 1 second where nothing is getting consumed, while partition 0 continues to receive records.

Let’s run some benchmarks to understand how serious this issue can be.

Illustrating the record_limit fetch wait issue with Dimster

Case 1 - Draining a giant backlog without producer load

Setup: 1 topic, 10 partitions, 5 consumers, max.poll.records=500 (the default), backlog size 400M records.

By the time the test reached the short test timeout, consumption was about 3,900 records/s, from a high of 1.2M records/s (no simulated processing time was configured). 98% of the 400M backlog drained in around 8 minutes.

The consumption slowdown started when lag was around 9M records. Extrapolating based on 3900 records/s, it should have taken 6 hours more to drain that 2% of the starting backlog.

Case 2 - Draining a skewed backlog

Setup: 1 topic, 12 partitions, 1 & 6 consumers, max.poll.records=500 (the default), backlog size 20M records.

Consumption starts strong but quickly drops to just shy of 2K records/s where it remained until the test reached the 20 minute drain timeout. Extrapolating, we can estimate a 2 hour drain time.

Case 3 - Skewed batch production workloads

Setup: 1 topic, 6 partitions, 1 consumer, max.poll.records=500 (the default), 6 brokers.

The consumer can’t handle the batch instantly, it needs time to process it. The consumption rate of the heaviest partition tops out at 1.5K records/s, building lag on that partition. In each of the three producer dumps, once the producer rate dropped to 0 and the 5 lightest partitions got drained, the heavy partition consumption rate crashed due to the fetch wait issue. Each consecutive production-batch increased the lag on the heaviest partition.

Case 4 - Backlogs with producer load

So far we’ve focus on backlog draining without producer load. But if producers keep going during the drain then the fetch wait issue can be mitigated. The size of the mitigation depends on the magnitude of producer rate.

Case 5 - Heavy skew + one-record-at-a-time fetches

If we reduce max.poll.records to 1, plus we have fewer consumers than partitions, plus we have serious skew, we encounter a double-whammy. Round-robin fetching that returns only a single record cannot prioritize the heavier partitions, in fact, the heavier partitions are penalized as the lighter partitions cause the consumer to spend most time fetching from them.

Mitigating these fetch wait cases

Primary mitigation: If you want to use record_limit, then the best mitigation is to use the partition count as the floor for consumer count. This completely side-steps the fetch wait problem and allows you to use record_limit without any risk of these weird performance issues under various types of skew.

Secondary mitigations (less effective or with drawbacks):

• Consider reducing fetch.max.wait.ms, if you have regular backlogs with no producer load (cases 1-3). The downside is if you get too aggressive, gone is long polling, instead you might hammer the Kafka brokers with a high frequency of fetch requests.
• Consider increasing max.poll.records if you experience case 5, as it allows the consumer to make up for the long periods between fetches to the heaviest partitions.
• Consider fixing your skew. However, even if your partitions are relatively balanced, if you accrue a very large backlog, then the lag can be skewed towards the end of the drain period.

How might we mitigate these pathological-fetch-waits with a change in how the Apache Kafka clients work?

1. Have clients not wait the full timeout period if the last fetch to that broker returned empty.
2. No round-robin fetch requests. Have the client send concurrent fetches to all brokers of the assigned partitions.
3. Have an additional communication channel between brokers and clients, so brokers can share lag information with clients.

I am sure this particular wrinkle with share groups will get worked out. In the mean time, the most sensible mitigation is to use the partition count as your floor for consumer count when using record_limit.