Module 03 · Day 3 · Kafka 4.3

Consumers

A consumer reads records and remembers how far it has read. A group of consumers shares the partitions of a topic. Most production problems with Kafka are consumer problems: lag, rebalances, and records processed twice or not at all.

Try it: a group shares six partitions

Each box is a partition. The label shows which consumer owns it. Add or remove a consumer and watch which partitions must stop.

Rebalance style

offset 0

What it is and why it exists

A consumer is the client library that reads records from Kafka. It asks a broker for records from a partition, starting at an offset. The broker does not push records and does not track what each consumer has seen. The consumer pulls, and the consumer keeps its own position.

One consumer is often too slow for a busy topic. So consumers join a consumer group, identified by group.id. Kafka gives each partition to exactly one consumer in the group. Two consumers in the same group never read the same partition at the same time. Two different groups each read everything, without affecting each other.

The group saves its position as a committed offset for each partition. After a crash or a restart, the next owner of the partition continues from the committed offset. When and how you commit decides if a failure loses records or repeats them.

The one-sentence version. A consumer group divides partitions among its members, each member pulls records and commits its position, and Kafka moves partitions to other members when one stops.

offset 1

Mental model

Think of a team that reads six notebooks. Each notebook has one reader at a time, and each reader keeps a bookmark in the notebook. If a reader leaves, someone else takes the notebook and continues from the bookmark. If the bookmark was not moved recently, the new reader reads some pages again.

The partition count is the limit of parallel work. Six partitions can feed at most six consumers in one group. A seventh has nothing to do.
The committed offset is the bookmark. It is the offset of the next record to read, not the last one read.
A rebalance is the moment partitions change owner. It happens when a member joins, leaves, or is judged dead.
Lag is the distance from the bookmark to the end of the notebook.

Three delivery guarantees

The order of two actions, "process the record" and "commit the offset", gives the guarantee.

Name	Order	After a crash
At most once	Commit, then process	Records that were committed but not processed are lost.
At least once	Process, then commit	Records that were processed but not committed are processed again.
Exactly once	Process and commit in one atomic step	Neither. This needs transactions (module 06) or a processing step that is safe to repeat.

Try it: crash in the middle of a batch

The consumer receives records 0 to 4. It crashes after it has processed three of them.

Commit

first run

after restart

Choose when to commit, then press the button.

offset 2

How it works internally

The poll loop

All consumer work happens inside one call that your code repeats: poll().

Join. On the first poll, the consumer finds the group's coordinator, a broker that manages this group, and joins the group.
Get an assignment. The consumer learns which partitions it owns.
Find the start. For each partition it asks for the committed offset. If there is none, auto.offset.reset decides: latest (the default) or earliest.
Fetch. It sends fetch requests to the leaders of its partitions. The broker answers when it has fetch.min.bytes of data or after fetch.max.wait.ms.
Return records. poll() hands your code up to max.poll.records records.
Process and commit. Your code processes them. The offset is committed by you, or by auto-commit during a later poll.

A consumer is not safe to use from several threads. One thread owns it and calls poll().

Where offsets are stored

Committed offsets are records in an internal topic, __consumer_offsets. It has 50 partitions and uses compaction, so it keeps the newest offset for each group, topic and partition. The coordinator for a group is the broker that leads one of those 50 partitions: the hash of the group id picks which one.

A group's offsets are deleted when the group has had no members for offsets.retention.minutes, 7 days by default. After that, the group starts again from auto.offset.reset.

How Kafka knows a consumer is alive

Check	Detects	Limit
Heartbeat	The process or its network is dead. A background thread sends heartbeats.	No heartbeat for the session timeout, 45 seconds.
Poll interval	The process is alive but your code is stuck or too slow.	No `poll()` call for `max.poll.interval.ms`, 5 minutes.

When either limit is passed, the consumer leaves the group and its partitions go to other members.

Two rebalance protocols

Kafka 4 has two ways to run a group. The consumer chooses with group.protocol.

	Classic (`group.protocol=classic`, the default)	New (`group.protocol=consumer`)
Who computes the assignment	One consumer in the group, chosen as leader, using `partition.assignment.strategy`.	The broker, using `group.remote.assignor`: `uniform` or `range`.
How a change happens	Every member must rejoin. With the older strategies, every member first gives up all its partitions.	The broker tells each member only what to give up and what to take. Members that are not affected continue.
Where timeouts are set	On the client: `session.timeout.ms`, `heartbeat.interval.ms`.	On the broker: `group.consumer.session.timeout.ms`, `group.consumer.heartbeat.interval.ms`. The two client settings are not supported.
Effect of one slow member	It can hold the whole rebalance.	It delays only its own partitions.

The demo at the top of this page shows the difference. "Stop everything" is the classic protocol with an older strategy such as RangeAssignor. "Move only what must move" is the new protocol, and also the classic protocol with CooperativeStickyAssignor.

Kafka 4.3 prints this line when a console consumer starts: "The consumer rebalance protocol (KIP-848) is production-ready! Set group.protocol=consumer to try it out."

Static membership

Each restart of a consumer normally causes two rebalances: one when it leaves and one when it returns. If you give each consumer a fixed name with group.instance.id, the coordinator keeps its partitions reserved until the session timeout. A restart that is faster than the timeout causes no rebalance. This helps rolling deployments.

Lag

Lag for a partition is the log end offset minus the committed offset. A steady small lag is normal. A lag that grows means the group processes slower than producers write. If the lag grows for longer than the retention time, the oldest unread records are deleted before the consumer reaches them.

Try it: how long until a slow consumer loses data?

Produced each second Consumed each second Retention

–lag change each hour

–how far behind after 24 hours

–time until unread data is deleted

offset 3

Every relevant config

Consumer defaults come from the Kafka 4.3 consumer configuration reference. Broker defaults were read from the lab cluster.

Consumer config	Default	What it does	When to change it
`group.id`	null	The name of the consumer group.	Always set it. One name for each application that needs its own copy of the data.
`group.protocol`	`classic`	Which rebalance protocol to use: `classic` or `consumer`.	Use `consumer` for new applications on Kafka 4.
`group.remote.assignor`	null	For the new protocol: the broker-side strategy. Null lets the broker choose; the lab cluster chose `uniform`.	Use `range` when two topics must have matching partitions on the same consumer, for joins.
`partition.assignment.strategy`	`RangeAssignor`, `CooperativeStickyAssignor`	For the classic protocol: the client-side strategy. The first in the list that all members support is used.	Put `CooperativeStickyAssignor` alone to avoid stop-everything rebalances.
`group.instance.id`	null	A fixed name for this consumer: static membership.	Set a stable, unique value for each instance to avoid rebalances on restart.
`enable.auto.commit`	`true`	Commits the position automatically during `poll()`.	Set `false` and commit yourself when losing a record is not acceptable.
`auto.commit.interval.ms`	5000 (5 seconds)	How often auto-commit runs.	Rarely.
`auto.offset.reset`	`latest`	Where to start when there is no committed offset, or it points to deleted data. Values: `latest`, `earliest`, `none`, or `by_duration:` followed by a duration.	Use `earliest` when a new group must process existing data. Use `none` to fail loudly.
`session.timeout.ms`	45000 (45 seconds)	Classic protocol only: time without a heartbeat before the member is removed.	Lower it to detect a dead consumer sooner. Too low gives false alarms.
`heartbeat.interval.ms`	3000 (3 seconds)	Classic protocol only: time between heartbeats.	Keep it at one third of the session timeout or lower.
`max.poll.interval.ms`	300000 (5 minutes)	Longest time allowed between two `poll()` calls.	Raise it, or lower `max.poll.records`, when processing one batch can take long.
`max.poll.records`	500	Most records returned by one `poll()`.	Lower it when each record takes long to process.
`fetch.min.bytes`	1	The broker waits until it has this much data before it answers.	Raise it for throughput and fewer requests, at the cost of delay.
`fetch.max.wait.ms`	500	Longest time the broker waits for `fetch.min.bytes`.	Lower it when a small delay matters.
`fetch.max.bytes`	52428800 (50 MiB)	Most data in one fetch answer.	Rarely.
`max.partition.fetch.bytes`	1048576 (1 MiB)	Most data from one partition in one fetch answer.	Raise it for large records or to drain a large lag faster.
`isolation.level`	`read_uncommitted`	Whether to see records from transactions that are not finished.	Set `read_committed` when producers use transactions. Module 06.
`client.rack`	empty	The consumer's location. It allows fetching from a nearby follower.	Set it in multi-zone clusters to cut cross-zone traffic.
`allow.auto.create.topics`	`true`	Lets a subscribe call create a missing topic, if the broker also allows it.	Set `false` so that a typing error does not create a topic.
`request.timeout.ms`, `default.api.timeout.ms`	30000, 60000	Time limits for one request and for one client call.	Rarely.
`check.crcs`	`true`	Verifies the checksum of each batch.	Leave it on.

Broker settings for groups

Broker config	Default	What it does
`group.coordinator.rebalance.protocols`	`classic,consumer,streams`	The protocols the broker accepts.
`group.consumer.session.timeout.ms`	45000	New protocol: time without a heartbeat before a member is removed.
`group.consumer.heartbeat.interval.ms`	5000	New protocol: the heartbeat interval the broker tells members to use.
`group.consumer.assignors`	`uniform,range`	New protocol: the strategies the broker offers. The first is the default.
`group.min.session.timeout.ms`, `group.max.session.timeout.ms`	6000, 1800000	Classic protocol: the range of session timeouts a client may ask for.
`group.initial.rebalance.delay.ms`	3000; the lab cluster sets 0	Classic protocol: wait for more members before the first rebalance of a new group. unverified default
`offsets.retention.minutes`	10080 (7 days)	How long the offsets of an empty group are kept.
`offsets.topic.num.partitions`	50	Partition count of `__consumer_offsets`. Do not change it after the cluster has started.

offset 4

Failure scenarios

A slow batch removes the consumer from the group

What breaks: The consumer processes a batch for longer than max.poll.interval.ms. It is removed, another member takes its partitions and processes the same records again. Then the first one fails to commit.
You observe: A commit error that says the group has rebalanced. Repeated records. Rebalances every few minutes.
Fix: Lower max.poll.records so that a batch is always fast. Or raise max.poll.interval.ms. Do not move slow work to another thread without also pausing the partition, or you lose the back-pressure.

Rebalance storm

What breaks: The group spends most of its time rebalancing and little time processing. Lag grows.
Cause: A rolling deployment restarts members one after another, or members keep passing the poll interval, or the session timeout is too low for a network with short pauses.
Fix: Use the new protocol or CooperativeStickyAssignor. Set group.instance.id for deployments. Fix the slow processing.

One bad record blocks a partition

What breaks: A record cannot be parsed or always fails in processing. The consumer throws, restarts, reads the same record, and throws again. This is a poison record.
You observe: Lag grows on one partition only. The committed offset does not move.
Fix: Catch the error, send the record to a separate topic for failed records (a dead-letter topic, module 08), commit, and continue. Decide in advance which errors deserve a retry.

Auto-commit loses records

What breaks: Your code hands records to a worker thread and calls poll() again. Auto-commit commits them. The process crashes before the worker has finished.
Result: Those records are never processed.
Fix: Set enable.auto.commit=false and commit only what is finished.

A new group skips existing data

What breaks: A new application starts and processes nothing, although the topic has data.
Cause: auto.offset.reset is latest by default, so a group without offsets starts at the end. The lab shows it: 6,000 records in the topic, 0 read.
Fix: Set earliest for applications that must process the history.

The offsets of an idle group disappear

What breaks: A job that runs once a month starts from the wrong place.
Cause: The group was empty for more than offsets.retention.minutes, 7 days. Its offsets were deleted.
Fix: Raise the broker setting, or store the position yourself, or keep one member running.

More consumers than partitions

What breaks: You scale the application from 6 to 12 instances and throughput does not change.
Cause: The topic has 6 partitions. Six instances own one each. The other six own nothing.
Fix: Add partitions (remember the ordering cost from module 01), or make each consumer faster, or use share groups (module 10) when order does not matter.

offset 5

Hands-on lab

Every command below was run on 4 October 2026 against the lab cluster. Outputs are real, with wide columns cut.

k() { docker exec -i kafka-1 /opt/kafka/bin/"$@"; }
B=kafka-1:19092
k kafka-topics.sh --bootstrap-server $B --create --topic events --partitions 6 --replication-factor 3
k kafka-producer-perf-test.sh --topic events --num-records 6000 --record-size 100 \
  --throughput -1 --command-property bootstrap.servers=$B

Step 1. A new group starts at the end

k kafka-console-consumer.sh --bootstrap-server $B --topic events --group g-latest --timeout-ms 5000
k kafka-consumer-groups.sh --bootstrap-server $B --describe --group g-latest

Processed a total of 0 messages

Consumer group 'g-latest' has no active members.
GROUP      TOPIC   PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
g-latest   events  0          1361            1361            0
g-latest   events  1          592             592             0
g-latest   events  2          934             934             0
g-latest   events  3          592             592             0
g-latest   events  4          1480            1480            0
g-latest   events  5          1041            1041            0

The topic has 6,000 records and the group read none. It committed the end of each partition. Notice also that the partitions are uneven: the records had no key, so they went in groups, as module 02 showed.

Step 2. Two consumers share the topic (classic protocol)

Start one consumer in each of two containers, in the background.

docker exec -d kafka-1 /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server $B \
  --topic events --group g-classic --from-beginning --command-property client.id=c1
docker exec -d kafka-2 /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server $B \
  --topic events --group g-classic --from-beginning --command-property client.id=c2

# wait 12 seconds
k kafka-consumer-groups.sh --bootstrap-server $B --describe --group g-classic
k kafka-consumer-groups.sh --bootstrap-server $B --describe --group g-classic --state

GROUP      TOPIC   PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG  CONSUMER-ID
g-classic  events  3          592             592             0    c2-442e6189-…
g-classic  events  4          1480            1480            0    c2-442e6189-…
g-classic  events  5          1041            1041            0    c2-442e6189-…
g-classic  events  0          1361            1361            0    c1-7b00d4f1-…
g-classic  events  1          592             592             0    c1-7b00d4f1-…
g-classic  events  2          934             934             0    c1-7b00d4f1-…

GROUP      COORDINATOR (ID)    ASSIGNMENT-STRATEGY  STATE   #MEMBERS
g-classic  kafka-1:19092 (1)   range                Stable  2

Each consumer owns three partitions. The strategy is range, which gives each member a continuous block: 0 to 2 and 3 to 5.

Step 3. The same with the new protocol

docker exec -d kafka-1 /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server $B \
  --topic events --group g-new --from-beginning \
  --command-property group.protocol=consumer --command-property client.id=n1
docker exec -d kafka-2 /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server $B \
  --topic events --group g-new --from-beginning \
  --command-property group.protocol=consumer --command-property client.id=n2

# wait 12 seconds
k kafka-consumer-groups.sh --bootstrap-server $B --describe --group g-new --state
k kafka-consumer-groups.sh --bootstrap-server $B --describe --group g-new --members

GROUP  COORDINATOR (ID)    ASSIGNMENT-STRATEGY  STATE   #MEMBERS
g-new  kafka-1:19092 (1)   uniform              Stable  2

GROUP  CONSUMER-ID             HOST         CLIENT-ID  #PARTITIONS
g-new  wIfVdQS2SLCMn4Mqa2HbTw  /172.23.0.2  n2         3
g-new  rB9ASMcrTDa1TxHsxFus2g  /172.23.0.4  n1         3

The strategy is now uniform, chosen by the broker. The member ids are short ids given by the broker, not the long client-made ids of the classic group.

Step 4. Stop one member

docker exec kafka-2 sh -c "pkill -f 'group g-classic'"
# wait 8 seconds
k kafka-consumer-groups.sh --bootstrap-server $B --describe --group g-classic --members

GROUP      CONSUMER-ID       HOST         CLIENT-ID  #PARTITIONS
g-classic  c1-7b00d4f1-…     /172.23.0.4  c1         6

The remaining consumer owns all six partitions. The stopped one left cleanly, so the group did not have to wait for the 45-second session timeout.

Step 5. Create lag and read it

docker exec kafka-1 sh -c "pkill -f ConsoleConsumer"; docker exec kafka-2 sh -c "pkill -f ConsoleConsumer"
k kafka-producer-perf-test.sh --topic events --num-records 3000 --record-size 100 \
  --throughput -1 --command-property bootstrap.servers=$B
k kafka-consumer-groups.sh --bootstrap-server $B --describe --group g-classic

Consumer group 'g-classic' has no active members.
GROUP      TOPIC   PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
g-classic  events  0          1361            1657            296
g-classic  events  1          592             928             336
g-classic  events  2          934             1230            296
g-classic  events  3          592             1184            592
g-classic  events  4          1480            2072            592
g-classic  events  5          1041            1929            888

The lag adds up to 3,000, the number of records written while no consumer ran. This command is the first thing to run when someone says "the consumer is behind".

Step 6. Move a group's position

A group must have no active members before you reset it. Always look at the plan first with --dry-run.

k kafka-consumer-groups.sh --bootstrap-server $B --group g-classic --topic events \
  --reset-offsets --to-earliest --dry-run
k kafka-consumer-groups.sh --bootstrap-server $B --group g-classic --topic events \
  --reset-offsets --to-earliest --execute

# go back 100 records on one partition of another group
k kafka-consumer-groups.sh --bootstrap-server $B --group g-new --topic events:0 \
  --reset-offsets --shift-by -100 --execute

GROUP      TOPIC   PARTITION  NEW-OFFSET
g-classic  events  3          0
g-classic  events  4          0
…
g-new      events  0          1261

After the first reset, partition 0 of g-classic shows a lag of 1,657: the whole partition. Other useful targets are --to-latest, --to-datetime and --to-offset.

Step 7. Look at the offsets topic

k kafka-topics.sh --bootstrap-server $B --describe --topic __consumer_offsets | head -1

Topic: __consumer_offsets  PartitionCount: 50  ReplicationFactor: 3
Configs: compression.type=producer,min.insync.replicas=2,cleanup.policy=compact,segment.bytes=10485760

Fifty partitions, compacted, with small segments so that the cleaner from module 01 can run often.

offset 6

Common mistakes and misconceptions

People think	What is true
Kafka pushes records to consumers.	Consumers pull. The consumer controls its own speed.
The committed offset is the last record processed.	It is the next record to read: the last processed offset plus one.
More consumers always means more throughput.	Only up to the partition count. Extra members of a group are idle.
Auto-commit gives at-least-once.	Only if your code finishes every record before the next `poll()`. With worker threads it can lose records.
Two groups share the records between them.	Each group gets every record. Sharing happens only inside one group.
A heartbeat proves the consumer is working.	It proves the process is alive. A stuck processing loop still sends heartbeats; the poll interval catches that.
Lag in records tells you how late you are.	It tells you how many records. How late depends on the write rate. Measure lag in time too.
A consumer can be shared between threads.	It cannot. Use one consumer for each thread, or one polling thread that hands work to others.

offset 7

Staff-level questions

Each record takes 2 seconds to process. How do you design the consumer?

First check the numbers: 500 records in one poll at 2 seconds each is 1,000 seconds, far past the 5-minute poll interval. So lower max.poll.records to a value that finishes well inside the interval, for example 50.

For more throughput, add partitions and consumers. If that is not enough, process records in a worker pool inside each consumer: pause the partition, hand out the records, call poll() to keep the membership alive, commit only the offset below which all records are finished, then resume. Keep records with the same key on the same worker if order matters.

How do you make processing safe to repeat?

At-least-once is the practical default, so repeats will happen. Give every event a unique id. In the database, write the result and the event id in one transaction, with a unique constraint on the id. A repeat then fails the constraint and is skipped.

Other options: make the operation naturally repeatable ("set balance to 50" instead of "add 10"), or store the Kafka offset together with the result and skip anything at or below it.

Should a new service use the classic or the new group protocol?

On Kafka 4 brokers with a 4.x Java client, use the new one: only affected partitions stop, a slow member cannot hold the group, and timeouts are managed in one place on the broker.

Check two things first. Clients in other languages may not support it yet. And a custom client-side assignor does not carry over, because the broker now computes the assignment.

What do you alert on for a consumer group?

Lag in time, not only in records, compared with what the business accepts and with the retention time. The rate of rebalances. The time since the last commit for each partition, which catches a poison record. And the number of members compared with the expected number.

A fixed threshold on record lag gives false alarms on busy topics and misses problems on quiet ones.

A deployment causes 10 minutes of duplicates each time. Why, and what do you change?

Each restart triggers rebalances. With an eager strategy, every member gives up its partitions; anything processed but not yet committed is processed again by the new owner.

Changes, in order of effect: commit when partitions are revoked; use the new protocol or the cooperative strategy so that only moved partitions are affected; set group.instance.id so that a quick restart is not a rebalance at all; and shut down cleanly so that the consumer commits and leaves.

When does reading from a follower make sense?

In a cluster that spans zones or regions, where traffic between zones costs money or time. With client.rack set on the consumer and the matching selector on the brokers, the consumer fetches from a follower in its own zone.

The cost is a small extra delay, because a follower learns the high watermark slightly after the leader.

offset 8

Self-test

1. A topic has 4 partitions and a group has 6 consumers. How many consumers receive records?

Four. Two are idle.

2. A consumer has processed offset 41. Which offset does it commit?

42: the next record to read.

3. Where are committed offsets stored?

In the compacted internal topic __consumer_offsets, which has 50 partitions.

4. A new group reads a topic that already has data and receives nothing. Why?

auto.offset.reset is latest by default, so the group starts at the end.

5. Which order of process and commit gives at-least-once?

Process first, then commit.

6. What does max.poll.interval.ms protect against?

A consumer that is alive but stuck or too slow: it sends heartbeats but does not call poll().

7. Name two differences between the classic and the new group protocol.

In the new protocol the broker computes the assignment, and only the affected partitions stop during a change. Session and heartbeat timeouts are set on the broker.

8. What does group.instance.id change?

The member becomes static. A restart that is shorter than the session timeout causes no rebalance.

9. Which command shows the lag of a group?

kafka-consumer-groups.sh --describe --group followed by the group name.

10. What must be true before you can reset a group's offsets with the tool?

The group has no active members.