Consumer Groups and Measuring Lag
How a consumer group divides a topic's partitions, what the committed offset really points at, and the one subtraction that turns broker metadata into the lag number an on-call engineer is paged about.
What is new here
sql-l5-system-design-reasoning introduced consumer lag as a vocabulary item: a number that goes up when a consumer falls behind. This lesson is not that lesson again. What is new is the exact formula, the metadata it is computed from, and the rebalancing mechanics that make the number jump for reasons that are not a slow consumer.
A group divides the partitions
A consumer group is a set of consumer instances that cooperate on one topic. The group's rule is short and it is the whole design: every partition is owned by exactly one member of the group at a time. Four partitions and two members means each member owns two. Four partitions and six members means two members sit idle, because a partition cannot be shared. That is the reason partition count is your throughput ceiling.
Different groups are completely independent. A fraud-detector group and an analytics-loader group both read payments from the beginning at their own pace, and neither can slow down or consume the other's records. That independence is what makes a log different from a queue.
The committed offset is the NEXT one to read
Each member periodically commits the offset it has reached, and the broker stores it per group and per partition. Read the definition carefully, because half of all Kafka bugs live in it: the committed offset is the offset the group will read NEXT after a restart, not the last offset it read. If a consumer has processed offsets 0 through 99, it commits 100.
Which gives the formula that this lesson drills:
lag = log_end_offset - committed_offset
That is it. It is the exact number kafka-consumer-groups --describe prints as LAG, computed from LOG-END-OFFSET minus CURRENT-OFFSET. A lag of zero means the consumer has read everything the broker has. Lag is per partition, and a group's lag is the sum across the partitions it owns.
| partition | owner | committed offset (next to read) | log end offset | lag |
|---|---|---|---|---|
| p0 | consumer A (fd-1) | 61000 | 92000 | 31000 |
| p1 | consumer A (fd-1) | 74600 | 88000 | 13400 |
| p2 | consumer B (fd-2) | 62000 | 84000 | 22000 |
Rebalancing: the pause you will be asked about
When a member joins, leaves, or dies, the group rebalances: partition ownership is reassigned across the surviving members. During the shuffle nobody owns the moving partitions, so processing on them pauses and the stored member_id is empty. A group that rebalances constantly, usually because a slow handler keeps blowing the poll timeout and getting evicted, can spend more time reassigning than consuming. Under the classic protocol the whole group stops during a rebalance, which is why it is called a stop-the-world rebalance, though the cooperative sticky assignor (KIP-429, Kafka 2.4) already made the classic protocol incremental for clients that opt in. KIP-848, generally available since Kafka 4.0, goes further and moves assignment from the group leader to the broker coordinator, so the join barrier disappears and the pause shrinks to the partitions that actually moved.
Common mistake: computing lag as log_end_offset - committed_offset - 1 because "the committed offset is the last one read". It is not. Off-by-one here is a real production bug: it makes a caught-up consumer report lag of -1 and it makes an alert threshold of zero fire forever.
Interview nuance: lag is a rate, not a number. A lag of 4 million that is shrinking every minute is a consumer recovering from a deploy and it needs nothing from you. A lag of 900 that is growing every minute is a consumer that will never catch up and it is the actual incident. The follow-up question is always "how do you fix it", and the answers are, in order: make the handler faster, add consumers up to the partition count, then add partitions. Notice that the third one is the only one that requires changing the topic.
On a real platform this differs. Here you join two small tables. On a real cluster the same join is done for you by
kafka-consumer-groups --describe --group <name>, which prints TOPIC, PARTITION, CURRENT-OFFSET, LOG-END-OFFSET, LAG, and CONSUMER-ID. In production nobody runs that by hand: Burrow or a JMX exporter scrapes it into Prometheus, and Amazon MSK publishesEstimatedMaxTimeLagandSumOffsetLagto CloudWatch. The metric to alert on is usually lag in SECONDS rather than in messages, because a lag of 10,000 means something very different on a topic doing 10 messages a second than on one doing 10,000.
CREATE TABLE topic_partitions (
topic TEXT,
partition_id INTEGER,
log_start_offset INTEGER, -- oldest offset still retained; rises as retention deletes from the head
log_end_offset INTEGER, -- the offset the NEXT produced record will receive
leader_broker INTEGER -- the broker serving reads and writes for this partition
);
INSERT INTO topic_partitions (topic, partition_id, log_start_offset, log_end_offset, leader_broker) VALUES
('orders', 0, 200000, 640000, 1),
('orders', 1, 210000, 630000, 2),
('orders', 2, 190000, 670000, 3),
('orders', 3, 205000, 665000, 1),
('orders', 4, 3400000, 5000000, 2),
('orders', 5, 215000, 625000, 3),
('clickstream', 0, 5120000, 5900000, 1),
('clickstream', 1, 5050000, 5860000, 2),
('clickstream', 2, 4980000, 5800000, 3),
('clickstream', 3, 5200000, 5990000, 1),
('clickstream', 4, 15200000, 23400000, 2),
('clickstream', 5, 5010000, 5780000, 3),
('clickstream', 6, 5300000, 6140000, 1),
('clickstream', 7, 5150000, 5910000, 2),
('clickstream', 8, 5080000, 5930000, 3),
('clickstream', 9, 4900000, 5730000, 1),
('clickstream', 10, 5240000, 6040000, 2),
('clickstream', 11, 5110000, 6020000, 3),
('sessions', 0, 3100000, 4000000, 1),
('sessions', 1, 2600000, 3300000, 2),
('sessions', 2, 2400000, 3000000, 3),
('sessions', 3, 2450000, 3050000, 1),
('payments', 0, 40000, 92000, 1),
('payments', 1, 41000, 88000, 2),
('payments', 2, 39000, 84000, 3),
('inventory_updates', 0, 5000, 130000, 2);
CREATE TABLE consumer_group_offsets (
group_id TEXT,
topic TEXT,
partition_id INTEGER,
committed_offset INTEGER, -- the NEXT offset this group will read, not the last one it read
last_commit_ts_ms INTEGER, -- epoch ms of the most recent commit for this partition
member_id TEXT -- the consumer instance that owns the partition; '' during a rebalance
);
INSERT INTO consumer_group_offsets (group_id, topic, partition_id, committed_offset, last_commit_ts_ms, member_id) VALUES
('fraud-detector', 'payments', 0, 61000, 1767288000000, 'fd-1'),
('fraud-detector', 'payments', 1, 74600, 1767288000000, 'fd-1'),
('fraud-detector', 'payments', 2, 62000, 1767288000000, 'fd-2'),
('order-enricher', 'orders', 0, 639800, 1767290390000, 'oe-1'),
('order-enricher', 'orders', 1, 629100, 1767290388000, 'oe-1'),
('order-enricher', 'orders', 2, 669650, 1767290389000, 'oe-2'),
('order-enricher', 'orders', 3, 665000, 1767289000000, 'oe-2'),
('order-enricher', 'orders', 4, 4995000, 1767290387000, ''),
('order-enricher', 'orders', 5, 625000, 1767290392000, ''),
('inventory-sync', 'inventory_updates', 0, 130000, 1767286800000, 'is-1');-- The same columns kafka-consumer-groups --describe prints, rebuilt from the two metadata tables.
SELECT o.topic,
o.partition_id,
o.committed_offset AS current_offset,
p.log_end_offset,
p.log_end_offset - o.committed_offset AS lag,
o.member_id
FROM consumer_group_offsets o
JOIN topic_partitions p ON p.topic = o.topic AND p.partition_id = o.partition_id
WHERE o.group_id = 'fraud-detector'
ORDER BY lag DESC;Apply
Your turn
The task this lesson builds to.
Write a query that returns the lag on every partition of payments for the fraud-detector group, as (partition_id, committed_offset, log_end_offset, lag), biggest lag first, over consumer_group_offsets(group_id, topic, partition_id, committed_offset, last_commit_ts_ms, member_id) joined to topic_partitions(topic, partition_id, log_start_offset, log_end_offset, leader_broker).
The two tables meet on both topic and partition_id, because a partition id alone is not unique across topics. Alias the computed column exactly lag.
3 hints and 1 automated check are waiting in the workspace.
Practice
Make it stick
A second problem on the same idea, plus 3 bonus drills.
Write a query that returns each consumer group with its total lag across all partitions and a stalled flag, as (group_id, total_lag, stalled), worst first, over the same two tables.
stalled is 1 when the group has lag AND has not committed anything in the last 15 minutes, and 0 otherwise. Treat the current time as 1767290400000 epoch milliseconds, and read commit times from last_commit_ts_ms. A fully caught-up group is never stalled, however old its commit is. Alias the columns exactly total_lag and stalled.
2 hints and 1 automated check are waiting in the workspace.