The keys in market-updates provide ordering per symbol. A consumer group divides the topic partitions between processes. This allows more consumers to share the load and recover partitions after a member fails.
I wanted to understand ownership changes while data continued to arrive. In my stock ticker experiment, I created a backlog and observed its removal. I then added a third consumer to see which partitions moved.
Partitions, not messages, get divided
A Kafka consumer group divides a topic's partitions among its members. Each partition is owned by exactly one consumer in the group at a time. With market-updates at 6 partitions:
- 2 consumers in the group: 3 partitions each.
- 3 consumers: 2 each.
- 6 consumers: 1 each. A 7th consumer has nothing left to own and sits idle. Partition count is the hard ceiling on useful parallelism.
That single owner rule is why per key order survives having multiple consumers at all. A given key, say BBCA, always hashes to the same partition, and that partition has exactly one owner at any moment. Every BBCA tick is handled by the same consumer, in the order it was produced. Ownership can move between consumers, but a partition's internal order never gets rearranged.
My test harness makes this observable by listening to KafkaJS's GROUP_JOIN event, which fires on every member each time the group settles on an assignment:
function makeConsumer(label: string, assignments: Assignments): Consumer {
const consumer = kafka.consumer({
groupId: GROUP,
sessionTimeout: 10000,
heartbeatInterval: 3000,
});
consumer.on(consumer.events.GROUP_JOIN, (e) => {
assignments.set(label, e.payload.memberAssignment);
});
return consumer;
}
The heartbeatInterval sets the heartbeat frequency. The coordinator uses sessionTimeout to detect a member that no longer sends heartbeats. A handler that blocks too long can cause the consumer to lose its assignment, even if the process remains alive.
The printed assignment gives three of the topic's six partitions to pod-1 and three to pod-2. It prints one line per member.
Committed offsets and what lag measures
Each partition stores records with increasing offsets. The log end offset identifies the next offset after the current log end. A consumer group commits the offset from which it should resume. After successful processing, this is normally the next record offset, not the offset of the last processed record.
The gap between those two numbers is lag:
lag = logEndOffset - committedOffset (per partition, summed across the topic)
Zero lag means the committed offsets have reached the partition ends used in the calculation. Stable positive lag means a backlog remains. Increasing lag means the backlog is growing. My helper calculates lag from broker metadata:
export async function lagWithAdmin(
admin: Admin,
groupId: string,
topic: string,
): Promise<LagReport> {
const [topicOffsets, groupOffsets] = await Promise.all([
admin.fetchTopicOffsets(topic),
admin.fetchOffsets({ groupId, topics: [topic] }),
]);
const committed = new Map<number, number>();
const entry = groupOffsets.find((g) => g.topic === topic);
for (const p of entry?.partitions ?? []) committed.set(p.partition, Number(p.offset));
const perPartition = topicOffsets.map((t) => {
const high = Number(t.high);
const low = Number(t.low);
let c = committed.get(t.partition) ?? -1;
if (c < 0) c = low; // group never committed here yet
return { partition: t.partition, high, committed: c, lag: Math.max(0, high - c) };
});
return { total: perPartition.reduce((s, p) => s + p.lag, 0), perPartition };
}
The c < 0 -> low fallback handles a partition with no committed group offset. It uses the earliest retained offset, low, instead of zero. This measures available history under the assumption that the group will start at that offset. The consumerLag() helper opens and closes its own admin connection.
Watching lag drain
To create a visible backlog, I produced 200,000 ticks across the 6 partitions before any consumer connected. The producer sent chunks of 10,000 ticks:
const CHUNK = 10000;
for (let sent = 0; sent < TOTAL; sent += CHUNK) {
const msgs = sim.batch(Math.min(CHUNK, TOTAL - sent)).map(tickToKafka);
await producer.send({ topic: TOPIC, messages: msgs });
}
Then pod-1 and pod-2 join the group. A short auto commit interval makes progress visible through committed offsets. A small delay in eachMessage lets me watch the backlog decrease:
await c.run({
autoCommitInterval: 400, // commit often so lag tracks real progress
eachMessage: async ({ message }) => {
const tick = tickFromKafka(message.value);
for (const rule of SAMPLE_ALERTS) evaluate(rule, tick);
counter.n++;
if (counter.n % 1000 === 0) await sleep(20); // gentle throttle
},
});
While that runs, I poll consumerLag() every 400ms and print a row per sample:
for (let i = 0; i < 25; i++) {
const { total } = await consumerLag(GROUP, TOPIC);
console.log(`t+${(i * 0.4).toFixed(1)}s lag=${total} consumed=${counter.n}`);
if (total === 0 && counter.n >= TOTAL) break;
await sleep(400);
}
The backlog starts with all 200,000 ticks unconsumed. Each consumer processes its 3 assigned partitions, and lag falls as committed offsets advance. The final commit can occur after counter.n reaches TOTAL. A falling lag curve shows progress, while flat or rising lag needs investigation.
Rebalancing: what happens when membership changes
Once lag has drained to zero, I add a third consumer, pod-3, to the same group and let it settle before reprinting the assignment:
const c3 = makeConsumer("pod-3", assignments);
await c3.connect();
await c3.subscribe({ topic: TOPIC, fromBeginning: true });
await c3.run({ eachMessage: async () => { counter.n++; } });
await sleep(3500);
printAssignments(assignments);
The subscribe and run calls add a member and trigger a rebalance. The group assignment changes from 3 partitions per consumer to 2 partitions per consumer. If a member disconnects or times out, its partitions move to the remaining members.
sequenceDiagram
participant Pod1 as pod-1
participant Pod2 as pod-2
participant Pod3 as pod-3
participant GC as "Group Coordinator"
Note over Pod1,Pod2: "Owns 3 partitions each"
Pod3->>GC: "Join the group"
GC->>Pod1: "Revoke partitions, pause fetching"
GC->>Pod2: "Revoke partitions, pause fetching"
GC->>GC: "Recompute assignment across 3 members"
GC->>Pod1: "New assignment: 2 partitions"
GC->>Pod2: "New assignment: 2 partitions"
GC->>Pod3: "New assignment: 2 partitions"
Note over Pod1,Pod3: "GROUP_JOIN fires per member, fetching resumes"
The diagram shows an "eager" rebalance. Existing members temporarily stop consumption while the group replaces the assignment. Cooperative or incremental protocols can limit the partitions that move, subject to broker and client support. Frequent rebalances interrupt useful work, so monitor membership changes and handler delays.
The ordering guarantee survives the churn
A rebalance changes partition ownership but preserves the stored record order. With unchanged partitioning, BBCA ticks continue to map to the same partition. The new consumer resumes from the committed offset, so it may repeat records that the previous consumer processed without a commit.
Redis Streams consumer groups look superficially similar from the outside. XREADGROUP also spreads work across a named group, but it has no partition concept underneath and no per key ownership at all. Any consumer can pick up any entry, including two entries for the same symbol at the same time.
Redis handles unfinished deliveries through the pending entries list. The next post compares that recovery model with Kafka partition reassignment.
Loading comments...