Kafka Share Groups - Pathological fetch waits with record_limit
In this post we’re going to see how 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. Kafka share groups have two methods of acquiring records: 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 , 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 , share consumers acquire records as slices of batches, where the size of the slice is determined by (now a strict cap). If you set 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 consistently did not complete explore-limits tests with 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. Fig 1. Dimster’s explore-limits mode regularly drains a backlog while searching for the highest sustainable rate under a target e2e latency 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 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 . 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 , fetches to brokers are round-robin if a consumer is assigned multiple partitions (on different brokers): Consumer sends a fetch to (which hosts partitions 0, 3, 6, 9) and gets back some records for partition 0. Poll is called again, triggering a fetch to (which hosts partitions 1, 4, 7), but there are no records. Poll is called again, triggering a fetch to (which hosts partitions 3, 5, 8) but there are no records. Poll is called again, triggering a fetch to , returning more records of partition 0. So what’s the problem? Can you see it yet? The problem is . 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. Fig 2. A single consumer, doing round-robin fetches across 3 brokers, does a lot of waiting when encountering brokers with 0 lag across their partitions (leader replicas). Let’s run some benchmarks to understand how serious this issue can be. Setup: 1 topic, 10 partitions, 5 consumers, max.poll.records=500 (the default), backlog size 400M records. This test generates a 400M record backlog using the , which generates a relatively balanced load across partitions. Each record is 50KB, resulting in a 20 GB backlog. The coordinator logs the drain progress: 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. What has happened is that due to some skew, half the partitions had drained causing the slow down. With 5 consumers and 10 partitions, each consumer was assigned two partitions, most likely on different brokers. So half of each consumer’s fetches were waiting for 500 ms and return nothing. The aggregate skew was relatively minor (the lightest partition had 39M and the heaviest had 45M), but the lag skew got worse as lighter partitions were drained. A 400M backlog is an extreme case. But we can trigger the slow down in much smaller backlogs if we use a more skewed message distributor mode. Let’s move onto case 2, where we’ll diagnose the pathological fetch wait problem further. Setup: 1 topic, 12 partitions, 1 & 6 consumers, max.poll.records=500 (the default), backlog size 20M records. To make this nasty, we’ll use a partition skew using Zipfian distribution with alpha=2. This is an extreme skew where the most heavily loaded partition (p0) will receive 64% of the records (12.8M), p1 will receive 16% (3.1M) and so on until p11 receives < 1% (88K). We’ll run two tests tests: with 1 consumer (assigned all 12 partitions) with 6 consumers (each assigned 2 partitions) Coordinator output excerpt: 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. Why just below 2000 records/s? A Prometheus query shows us the lower loaded partitions drained quickly and that the slow down in aggregate consumption correlated to an interval where p2 and p3 finished draining and p0 and p1 consumption dropped massively at the same time. Fig 3. Showing when each partition got drained, and the impact on consumption of the heavy partitions p0 and p1. Inspecting the partitions, we see that the leader of p0 is hosted by broker-1 and p1 by broker-2. So we’ve hit the following scenario (where only p0 and p1 have remaining lag): Fig 4. The topology in this one-consumer test. In a one second period (with zero ms processing time): Fetch 500 from Fetch 500 from Fetch 0 from with 500 ms fetch latency Fetch 500 from Fetch 500 from Fetch 0 from with 500 ms fetch latency In one second, the consumer can do two rounds of fetching from p0 and p1 (500 at a time), though in reality there is some overhead, which matches the 1990 records/s. This test did no better. In fact, the slow down happened far earlier. Fig 5. The drain-rate of the one-consumer and six-consumer tests. The six-consumer test hit the slowdown far earlier. We can see, while the test runs, that partitions p0-p4 still have lag (proportional to the skew). Inspecting the partition placements and share group assignments, we that these 5 partitions with lag are spread across 5 consumers. Each of these 5 consumers is assigned one heavy partition (still with lag) and one lighter partition (no drained). Fig 6. The toplogy and fetch waits in the six-consumer test. With zero processing time in this test, in a one second period, each of the 5 degraded consumers would: Fetch 500 records from the heavy partition Block on the light partition for 500ms Fetch 500 records from the heavy partition Block on the light partition for 500ms. The slow down happened earlier as each consumer fetched only from one partition per broker, whereas the single consumer fetched from 4 partitions per broker, so took longer to completely drain entire brokers. You might be thinking, how often do we need to drain a backlog, all the while the producer rate is 0? Let’s move onto case 3. Setup: 1 topic, 6 partitions, 1 consumer, max.poll.records=500 (the default), 6 brokers. One such case of draining backlogs without producer load is that of workloads where producers periodically dump a large batch of records in a topic. In between each dump there are no incoming records at all. We can model this with Dimster using its workfield. Fig 7. Per-partition rates + aggregate lag. Note that due to the rate calculated over a 1 minute interval, the short peak of 10,000 records/s is not shown. 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. In this test I used 6 brokers, to ensure that each partition was on a separate broker, to exacerbate the problem. Obvious, this test doesn’t need 6 brokers, but in production you might run 6 brokers or 12 brokers or more. In such clusters, it would be the norm for the leader replicas of a topic to be not be co-located on the same brokers. 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. If a record arrives at a light partition twice per second, then the fetch wait issue may not be mitigated at all. The following chart shows a small backlog from one cycle of a batch-production workload. After the peak of 12,000/s, the producer rate drops to 0 for three minutes, then every two minutes increases until it finally reaches 900 records per second across all partitions (with 64% going to p0). Fig 8. Demonstrating how the producer rate can affect the consumer rate. We can see that as the producer rate increases, the drain rate of the single partition backlog accelerates. The producer rate accelerates the consumer rate. This tells us that a continued producer rate may or may not mitigate the fetch wait issue. The lower rate, the less effective the mitigation. If we reduce 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. In the worst case, the consumer spends 500 ms waiting for a fetch to a lighter partition, but comes up empty, while the heavier partition is filling up. One such case, designed to maximize this pathology is: 6 records per second 12 partitions max.poll.records=1 average processing time is 10 ms 4 different load skews (via workload field): : Almost perfect uniform distribution : Light skew. . High skew : With one producer -> high temporal skew, low aggregate skew. Why high temporal skew for ? Basically, the single producer chooses a partition and sends records to it for a while, then switches to another partition for a while and so on. Within a short period of time, only one partition is receiving records. You can see the partition skew of these four tests below: Fig 9. The partition skews of each test (PINNED_PARTITIONS, KEY_ROUND_ROBIN, PARTITION_ZIPF, NO_KEY). The results show that the Zipfian 1.5 test reached only 2 records/s with one consumer and 5 records/s with six consumers. The test also saw elevated end-to-end latency, though it did not continue to grow. Fig 10. Consumer rate and p99 e2e latency over time, of the four tests. Primary mitigation : If you want to use , 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 without any risk of these weird performance issues under various types of skew. Secondary mitigations (less effective or with drawbacks): Consider reducing , 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 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. The following chart shows drain times for a high skew backlog with different with 6 consumers and no producer load: How might we mitigate these pathological-fetch-waits with a change in how the Apache Kafka clients work? Have clients not wait the full timeout period if the last fetch to that broker returned empty. This would help in backlog drain scenarios without producer load, but not low producer load (where fetches are non-empty but high latency). No round-robin fetch requests. Have the client send concurrent fetches to all brokers of the assigned partitions. However, this weakens one of the main objectives on , which is to place a hard cap on the number of records inflight for a given consumer, in order to avoid reaching record timeouts and redelivery. Have an additional communication channel between brokers and clients, so brokers can share lag information with clients (so clients can preferentially fetch from higher lag partitions). 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 . fewer consumers than partitions and various cases of “partition skew” Consumer sends a fetch to (which hosts partitions 0, 3, 6, 9) and gets back some records for partition 0. Poll is called again, triggering a fetch to (which hosts partitions 1, 4, 7), but there are no records. Poll is called again, triggering a fetch to (which hosts partitions 3, 5, 8) but there are no records. Poll is called again, triggering a fetch to , returning more records of partition 0. with 1 consumer (assigned all 12 partitions) with 6 consumers (each assigned 2 partitions) Fetch 500 from Fetch 500 from Fetch 0 from with 500 ms fetch latency Fetch 500 from Fetch 500 from Fetch 0 from with 500 ms fetch latency Fetch 500 records from the heavy partition Block on the light partition for 500ms Fetch 500 records from the heavy partition Block on the light partition for 500ms. 6 records per second 12 partitions max.poll.records=1 average processing time is 10 ms 4 different load skews (via workload field): : Almost perfect uniform distribution : Light skew. . High skew : With one producer -> high temporal skew, low aggregate skew. Consider reducing , 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 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. Have clients not wait the full timeout period if the last fetch to that broker returned empty. This would help in backlog drain scenarios without producer load, but not low producer load (where fetches are non-empty but high latency). No round-robin fetch requests. Have the client send concurrent fetches to all brokers of the assigned partitions. However, this weakens one of the main objectives on , which is to place a hard cap on the number of records inflight for a given consumer, in order to avoid reaching record timeouts and redelivery. Have an additional communication channel between brokers and clients, so brokers can share lag information with clients (so clients can preferentially fetch from higher lag partitions).
Backend
API
Performance
AI
Database
Testing
Ruby
Frontend
Web Development
Programming
C
Shell
Go
Open Source
Hardware
JavaScript
CSS
Web
DevOps
Linux
Cloud
C++