Kafka Consumer Groups and Offset Management

Data Engineering & Modern Data Platforms · Lesson 40 · Real-Time Data Engineering and Kafka

Lesson Overview

The landing loop in Kafka Producers and Consumers calls commitSync after a durable write and uses the group id learning-progress-raw-landing. That call is the entire recovery story for landing, and it is easy to misread. A committed offset is not the last record you finished. It is the next record the group should read. A second program that uses the same group id is not a second subscriber. It is another member of the same assignment, and it will take partitions away from landing. The activity tile, which looks like a subscriber, does not commit this group's offsets at all. Its cursor is the checkpoint from Structured Streaming with Spark. An incident commander who "resets the consumer" will move one of these and not the other, or will move the wrong one because both are called offsets in conversation.

This lesson is the group protocol from the data engineer's side: how partitions are assigned, when a rebalance runs, what commitSync writes into __consumer_offsets, how lag is computed, and how a rewind is done with the group stopped. It is also the place we separate that machinery from Spark. The primer's Consumer Groups, Offset Management, Rebalancing, Manual Commit, and Auto Commit pages are the protocol catalog. Use them for assignor edge cases. Use this lesson to keep the landing prefix, the quality auditor, and the tile from sharing a cursor they do not share.

Learning Objectives

By the end of this lesson you will be able to:

1. Explain a consumer group as one assignment of partitions, and give each product its own group.id.
2. Define a committed offset as the next record to read, stored for the group in __consumer_offsets.
3. Commit after a durable side effect with commitSync, and say what auto-commit commits instead.
4. Describe an eager rebalance and a cooperative one, and set CooperativeStickyAssignor when every member can use it.
5. Separate max.poll.interval.ms from session.timeout.ms, and name CommitFailedException after a revoked assignment.
6. Compute lag from the describe output, and rewind a stopped group to a known offset without touching the Spark checkpoint.
7. Refuse to treat the tile's checkpoint as a group offset, and refuse to let a notebook join the landing group.

Prerequisites

• Kafka Producers and Consumers: the poll loop, enable.auto.commit=false, and the offset-named raw object.
• Kafka Topics, Partitions and Replication: twelve progress partitions, so at most twelve landing members do useful work.
• Structured Streaming with Spark: checkpoint directory, startingOffsets, failOnDataLoss.
• KafkaConsumer in kafka-clients: commitSync, commitAsync, ConsumerRebalanceListener, ConsumerGroupMetadata.
• The distribution tool kafka-consumer-groups.sh for describe and for a stopped-group rewind. /learn/kafka/consumer-lag is the primer on reading that output under load.

Detailed Core Concepts

One group is one cursor

A consumer group is a named set of members that share one assignment of the partitions of the topics they subscribe to. For learning.progress with twelve partitions, the group learning-progress-raw-landing may have one member owning all twelve, or three members owning four each, or twelve members owning one each. A thirteenth member sits idle. The group coordinator, which is a broker, decides membership and assignment. Members send heartbeats and poll. The coordinator does not know whether you fsynced a file. It knows generations, assignments, and the offsets you committed.

A second group, learning-progress-quality, subscribes to the same topic and has its own assignment and its own offsets. That is fan-out. Putting the auditor in the landing group is not fan-out. It is a stolen partition. The tile does not join either group to recover. If a Spark source configuration happens to set a group.id for metrics, that id is still not the resume cursor. Resume is the checkpoint. Write the three names in the catalog and reject a fourth process that reuses any of them.

The number you commit is the next offset

Suppose partition 4 returns records at offsets 100, 101, and 102. The consumer has not committed. The position to commit after you finish them is 103. On restart the member fetches 103. If you commit 102, you will read 102 again only if 102 meant "next," which it does not. The Apache KafkaConsumer contract is explicit: committed offsets are the next message to read. commitSync() with no arguments commits that next offset for every partition returned by the last poll. You do not add one yourself when you use that overload. You do add one when you build an OffsetAndMetadata by hand from record.offset().

javaCopy
offsets.put(
    new TopicPartition(record.topic(), record.partition()),
    new OffsetAndMetadata(record.offset() + 1));

The + 1 is the difference between "I have this record" and "give me what comes after it." Forgetting it replays the last record forever if your object write is not idempotent, or once more if it is. Adding it twice skips a record. Prefer commitSync() with no map when the whole poll batch succeeded, so the client does the arithmetic from the records it returned.

The commit is a produce to the internal compacted topic __consumer_offsets, keyed by group, topic, and partition. It is not a flag on the progress log. Retention and compaction of the internal topic are platform concerns. If commits succeed and a restart reads from the earliest progress offset anyway, you are looking at a different group id, a reset, or a group with no commit yet and auto.offset.reset=earliest. You are not looking at a broken progress replica.

Auto-commit is a timer, not a sink acknowledgement

enable.auto.commit=true records, on the cadence of auto.commit.interval.ms (default five seconds), the offsets of records the client has already returned from poll. The commit happens inside a later poll or during close. It does not happen when your object store returns. If you poll, start a slow put, and the next poll is what triggers the commit, you can commit offsets whose put is still in flight. A crash then skips them. That is at-most-once behavior hiding in a default. If you never call poll again because the put is still running, the auto-commit may not have fired, and a crash redelivers. Which one you get depends on timing. Landing cannot depend on timing. The landing consumer sets enable.auto.commit=false and calls commitSync after syncRaw and after any DLQ produce has completed.

commitAsync returns before the commit is known. It is appropriate in a tight loop that also commitSyncs in the rebalance listener and on shutdown. A failure callback on the async commit means the position may not be stored. The next records are already being processed. For the landing volume of this bus, synchronous commits after each batch are simpler and the latency is a round trip to the coordinator, not a round trip through the lake. Switch to async only when that round trip shows up in lag and you have the revoke-path commitSync in place. Do not use async because an example did.

Rebalance: eager, cooperative, and the poll interval

A rebalance runs when a member joins, leaves, fails, or when the subscription changes. The coordinator bumps the generation and assigns partitions under the strategy the members have in common.

RangeAssignor is an eager protocol. Members revoke all partitions, stop, then receive a new assignment. During that gap nobody consumes those partitions. A deploy of three landing instances, one after another, can rebalance three times and reshuffle partitions that did not need to move.

CooperativeStickyAssignor revokes only the partitions that must move. Members keep consuming the rest. That is the strategy this landing group should run once every member is on a client that supports it. The configuration is partition.assignment.strategy. On current Apache clients the default list starts with RangeAssignor, and the group picks the first strategy every member supports. A group that leaves the default therefore stays eager even though the cooperative class is on the classpath. Set the value to org.apache.kafka.clients.consumer.CooperativeStickyAssignor on every landing member in one coordinated deploy. A mixed group, some members listing only the cooperative assignor and some listing only range, will not stabilize on a strategy. Roll the config so every member's list includes the cooperative assignor first, and only then drop the range assignor.

Two timeouts are not the same failure. heartbeat.interval.ms and session.timeout.ms are the background heartbeat. Since KIP-62 the heartbeat thread is separate from processing. A member whose heartbeats stop for session.timeout.ms (default 45 seconds) is ejected. max.poll.interval.ms (default five minutes) is the longest gap allowed between poll calls. A member that heartbeats and does not poll is treated as stuck, because the coordinator believes the processing thread is wedged. A lake put that blocks the poll thread for six minutes trips this even though heartbeats are fine. The exception you then see on commitSync is CommitFailedException: the generation moved on, the partitions were given away, and a commit from the old member must not overwrite the new member's progress.

The landing response to CommitFailedException is to stop and look. Do not catch it and continue writing. Your writes may have landed for a partition you no longer own, and the new owner is also writing. Idempotent object names keep that from duplicating bytes. A non-idempotent side sink will double. After the exception, fix the pause: smaller max.poll.records, a faster put, or a higher interval that still matches a real bound. An interval of an hour means a stuck member holds partitions for an hour. That is a long stall of the morning handoff. Five to ten minutes is a reasonable ceiling for this job if puts are healthy inside it.

group.instance.id is static membership. A process that restarts with the same instance id, within session.timeout.ms, receives the same partitions without a group-wide rebalance. That fits a landing deploy if the restart is quicker than the session timeout you set. A session timeout long enough for a slow restart is also a slower detection of a real crash. Use static membership when deploys are causing needless reshuffles and you have measured the restart time. Do not set the session timeout to an hour to avoid learning the assignor.

Listeners: commit on revoke, do not commit on lost

ConsumerRebalanceListener.onPartitionsRevoked runs before partitions are taken, in the cooperative case only for the partitions that are moving. Commit the work you finished for those partitions. Do not start a new poll from inside the listener.

onPartitionsLost runs when the member was already ejected, for example after a session timeout. The partitions may belong to someone else. Do not commit. A commit can fail, or it can race the new owner. Close resources for those partitions and let the new member read from the last committed offset. The crash window is the same one lesson 38 made idempotent: finished objects may exist for offsets that will be read again.

onPartitionsAssigned is where you open writers for the partitions you now own. It is not where you seek to earliest. Seeking to earliest on every assignment rewrites the prefix from retention's start on every deploy. The committed offset is the seek the client already does when you do not override it.

Lag

Lag for a partition is the log end offset minus the committed offset. Because the committed value is the next record, a consumer that has fetched and not committed still shows that fetch in the lag. That is the lag you want for landing: bytes not yet safe in the group sense. A consumer that is caught up and idle shows lag zero or a small number equal to records published since the last commit. Describe the group:

bashCopy
kafka-consumer-groups.sh --bootstrap-server kafka1.internal:9092 \
  --describe --group learning-progress-raw-landing

The columns that matter are partition, current offset, log end offset, and lag. Add the member assignment (--members --verbose) when you suspect a laptop in the group. Alert on lag per partition above a bound you can recover inside retention, and on a lag that grows for longer than the morning handoff allows. A topic-level sum hides one stuck partition. Twelve partitions with lag zero and one with lag equal to the holiday week is a hole in one slice of course_day.

The tile's lag is not this number. Spark reports input rate and the offsets in the checkpoint. An alert only on the landing group will stay green while the tile is stopped. An alert only on the tile will stay green while landing is stopped and the morning table is about to miss a day. Both pages exist. They do not share a threshold, because their cursors do not share a commit.

Reset is an administrative write

kafka-consumer-groups.sh --reset-offsets changes the committed next-offset. It is the rewind and the skip. Stop the members first. A reset against a live group fails or races a commit, depending on the tool version, and either outcome is a bad afternoon. Dry-run without --execute, read the plan, then execute.

Rewind partition 4 to offset 1000 only if the raw objects at and after 1000 can be replaced by the idempotent writer. The command is by partition because offsets are. Resetting the whole topic to earliest replays retention into the lake. That is a rebuild, not a tweak, and it does not move the Spark checkpoint. Resetting to latest skips the gap and makes the hole permanent once retention deletes it. Record the decision in the incident. auto.offset.reset does not apply to a group that already has commits. Changing it from earliest to latest in a config file does not rewind or skip. People learn that during incidents and then reach for --reset-offsets in a hurry. The dry-run is the pause.

auto.offset.reset=none fails the consumer if there is no committed offset. That is a good setting for a group that must never silently start at latest or earliest, such as a financial consumer. Landing uses earliest so a newly created group catches up the retained log on purpose. The quality auditor can use latest if its product definition is "from now," and that choice is written down so nobody expects it to backfill.

What the group will not do

It will not commit the Spark checkpoint. It will not dedupe event_id. It will not hold a partition open across a poll interval you exceeded. It will not remember a laptop's intent. It will not keep offsets for a group id you renamed. A rename is a new group. The old commits stay under the old id until the internal topic compacts them away, and the new group follows auto.offset.reset. Renaming learning-progress-raw-landing to "v2" without a reset plan starts at earliest or latest and calls it a deploy.

Architecture and Design Diagrams

mermaidCopy
flowchart TB
    topic[learning.progress 12 partitions]
    topic --> g1[Group learning-progress-raw-landing]
    topic --> g2[Group learning-progress-quality]
    topic --> spark[Spark checkpoint separate store]
    g1 --> off1[Commits in __consumer_offsets]
    g2 --> off2[Different commits same internal topic]
    spark --> files[Checkpoint offset and commit logs]
    g1 --> raw[Raw objects named by offset]
    g2 --> metrics[Quality counts]
    spark --> tile[Five-minute tile]

Three readers, three cursors, one log. __consumer_offsets holds the two group commits. It does not hold the checkpoint.

mermaidCopy
sequenceDiagram
    participant M1 as Landing member A
    participant C as Group coordinator
    participant M2 as Landing member B

    M1->>C: poll and heartbeat generation N
    M2->>C: join
    C->>M1: revoke partitions that move
    Note over M1: commitSync finished offsets for revoked partitions
    C->>M2: assign those partitions at committed next offset
    M2->>M2: poll continues
    Note over M1: partitions that did not move stay with A on a cooperative assignor

An eager assignor would revoke everything from A before the new assignment. The landing prefix should not depend on which one you run for correctness. It will depend on it for how long the morning handoff stalls during a deploy.

mermaidCopy
flowchart LR
    poll[poll returns offsets 100 to 102]
    work[fsync object and DLQ get]
    commit[commitSync next offset 103]
    crash[crash before commit]
    poll --> work --> commit
    work --> crash
    crash --> replay[next poll reads 100 again]
    replay --> same[same object name replaced]

The replay arrow is safe because of the object name. Remove the object name and the replay arrow is a duplicate in course_day.

Technical Explanation

Assignment and the twelve-partition ceiling

The landing deployment runs one KafkaConsumer per process, on the poll thread from lesson 38. Three processes are three members. The cooperative sticky assignor keeps their partitions stable across a fourth member joining, moving only the partitions the new member takes. Range assignment can swap partition sets between members even when the set of members is similar, which shows up as a burst of replaced objects and a confused lag chart. Sticky behavior is the point of the assignor, not a cosmetic.

You cannot get more than twelve-way parallelism on this topic without more partitions, and more partitions are the contract change in lesson 39. A thread pool inside one member does not create a thirteenth assignment. If you hand records to worker threads, the poll thread must still poll inside max.poll.interval.ms, and commits must happen only for records whose workers have finished, on the poll thread. That design is easy to get wrong. Prefer more members over a pool until the pool is the measured requirement. The primer's Parallel Consumers page discusses the pattern. The bus does not need it at twelve partitions and a bounded put.

The revoke listener that matches the loop

javaCopy
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.common.TopicPartition;

import java.util.Collection;

public final class LandingRebalance implements ConsumerRebalanceListener {
    private final Runnable commitFinished;

    public LandingRebalance(Runnable commitFinished) {
        this.commitFinished = commitFinished;
    }

    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        // Only partitions we finished writing. commitSync the consumer's current position
        // for them before they move. The runnable calls consumer.commitSync().
        commitFinished.run();
    }

    @Override
    public void onPartitionsLost(Collection<TopicPartition> partitions) {
        // Membership already lost. Do not commit. Idempotent objects absorb the replay.
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        // Open writers. Do not seek to earliest.
    }
}

Wire it as consumer.subscribe(List.of("learning.progress"), listener). The runnable must call commitSync on the same consumer on the poll thread. The listener runs on that thread. Do not spawn a commit on a side thread from onPartitionsRevoked.

If commitSync inside revoke throws CommitFailedException, the generation is already gone. Log it and leave. The new owner will replay from the last successful commit. This is the path static membership and a short restart are meant to make rare, not a path you paper over with a retry loop that blocks the rebalance.

Inspect and rewind

Describe is safe while the group is running. Reset is not.

bashCopy
# Plan a rewind of one partition. Group members must be stopped before --execute.
kafka-consumer-groups.sh --bootstrap-server kafka1.internal:9092 \
  --group learning-progress-raw-landing \
  --topic learning.progress:4 \
  --reset-offsets --to-offset 1000 \
  --dry-run

--to-offset 1000 sets the next record to 1000. The record at 1000 will be read again. It does not mean "I finished 1000." After you agree with the plan, stop the group, run with --execute instead of --dry-run, confirm describe shows current offset 1000 and a lag, then start the members. The Spark checkpoint is not an argument to this command. If the tile also needs to move, that is a checkpoint operation in the Spark runbook, decided separately.

--to-datetime uses the broker timestamp index, which is CreateTime or LogAppendTime, not payload event_time. A rewind to "yesterday at 18:00 event time" is not this flag. Find the offset in the raw zone or by scanning, then --to-offset. Using datetime and hoping it matches event_time reintroduces the clock bug from lesson 36.

Quality group

The auditor group uses a different group.id, its own client.id, and auto.offset.reset according to its product. It may lag. It must not produce to the landing DLQ under the landing principal's identity in a way that commits the landing group. Give it read on the topic and write only on topics it owns. Its lag alert is a quality freshness alert, not a course_day completeness alert. Completeness is the landing group's committed offset versus the log end, plus the manifest check in the morning job.

Spark, restated as an operational rule

When the tile is late, open the checkpoint and the streaming query progress. When landing is late, open the group describe. When both are late, the broker or the network is the shared cause and both cursors will move again from where they are once the cluster is healthy. Do not align them by resetting both to the same number. The tile may have committed a micro-batch the landing object does not have yet, or the reverse. Alignment is a coincidence, not an invariant.

startingOffsets on the Spark source applies when the checkpoint is empty. auto.offset.reset applies when the group has no commit. Teaching operators one phrase, "start at latest," for both systems is how a recovered job skips a day and a recovered group skips a day in the same incident. The runbook names the flag that belongs to the cursor.

Practical Implementation or Design Walkthrough

Problem

Run two landing members in learning-progress-raw-landing with the cooperative assignor, commit on revoke, and prove a third process in a different group can read the same topic without moving landing's offsets. Then rewind one partition of the landing group with the members stopped, and show the Spark checkpoint path is absent from the command.

Configuration added to the lesson 38 consumer

javaCopy
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.CooperativeStickyAssignor;

props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
        CooperativeStickyAssignor.class.getName());
props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300_000);
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 45_000);
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

MAX_POLL_INTERVAL_MS_CONFIG is set explicitly at the default so a later edit is a visible diff. Each member sets a distinct client.id (progress-raw-landing-a, progress-raw-landing-b) and, if you enable static membership, a stable group.instance.id equal to the host role, not a random UUID per process start. A random instance id on every restart is static membership that never matches, plus a session timeout that now waits to notice the old id.

Steps

1. Start member A alone. Describe the group. Confirm one member and twelve partitions assigned. Publish a valid event. Confirm lag returns to zero after commit and the object exists.
2. Start member B. Confirm a rebalance, both members assigned, and the sum of assigned partitions still twelve. Confirm objects written during the join are not duplicated for a given offset.
3. Start a quality consumer with group.id=learning-progress-quality. Confirm landing's current offsets do not jump when the quality group polls. Describe both groups. The current-offset columns are allowed to differ.
4. Block the poll thread in a lab build for longer than max.poll.interval.ms. Observe the rebalance and CommitFailedException on the subsequent commit. Restore the loop. Confirm the other member or the restarted member continues from the last successful commit.
5. Stop A and B. Dry-run a reset of one partition --to-offset equal to an offset you already landed. Execute. Start the members. Confirm the object for that offset is replaced and the line count for that offset is one.
6. Write the runbook line: the tile checkpoint path, and the statement that this reset did not change it.

Expected evidence

Two describe transcripts (one member, then two), the quality group's separate current offset, the exception class from the blocked poll, and the dry-run plan plus the post-reset object listing. A screenshot of a lag dashboard without the group id is not evidence.

Enterprise Real-World Scenarios

Scenario 1: The morning table is short and landing lag is not zero

At 05:40 UTC the course_day completeness check compares manifests with log end offsets. Partition 7 of learning-progress-raw-landing shows lag of several million. The other partitions are near zero. The tile is current, because its checkpoint advanced on every partition overnight. Product asks to "copy the tile into the morning table." The tile has dropped late event time and is the wrong definition, and it also cannot fill a raw prefix the morning job reads.

The stuck member owns partition 7 and has been exceeding max.poll.interval.ms because a single object put hung. CommitFailedException was caught and ignored in a recent change, so the process looks alive and writes nothing new. The fix is to fail the process on that exception, restart the member, and let it resume from the last commit. Idempotent objects make the overlapping write safe. The morning job waits or publishes a partial day with an explicit gap flag. It does not read the tile. Retention of fourteen days means this lag is recoverable at 05:40. The same lag discovered after retention would not be. The alert threshold has to fire while the offsets exist, which is an operations cost you pay in pages so you do not pay it in missing course-days.

Security shows up if the hung put is an expired lake credential. The group protocol is healthy and the side effect is not. Lag is the symptom either way. The runbook's first split is "describe the group" versus "read the put error."

Scenario 2: A deploy reshuffles with the range assignor

Landing runs three members and the default assignor. A deploy restarts them one at a time. Each restart revokes every partition. For two minutes per restart the raw prefix stalls. The tile, on its own checkpoint, keeps up. A freshness alert on landing pages three times in one deploy and then clears. Someone raises max.poll.interval.ms to thirty minutes to "stop the flaps." The flaps are rebalances, not poll gaps. The new interval now hides a truly stuck put.

The change that matches the symptom is CooperativeStickyAssignor on all three members, plus static membership if restarts are shorter than the session timeout. The deploy then revokes only the partitions of the member that is down. Lag blips on those partitions and nowhere else. The poll interval stays at five minutes so a hung put still fails the member. Cost is a coordinated config change, not more brokers. The tile's checkpoint is untouched, which is the regression test: tile lag must not move because landing changed assignors.

Scenario 3: Rewind requested for a bad validator

Yesterday's landing wrote objects whose progress_pct of 140 was accepted, because classify had a bug. The validator is fixed. You need to reprocess from yesterday 00:00 UTC broker time, which you have mapped to an offset per partition by scanning the raw lines' kafka_timestamp, not by --to-datetime alone if you are unsure of timestamp type. You stop the landing group, reset each partition to the mapped offset, and start. Objects are replaced. The quality group is left where it is, or reset on purpose if its counts must be rebuilt. The tile is left alone if its windows were wrong for the same bug: that is a checkpoint and a sink rewrite, a different decision, because reprocessing the tile changes closed windows in Parquet. One incident, three cursors, three decisions. The scenario fails if one --reset-offsets --to-earliest is used for "everything."

Best Practices

• One product, one group.id. The landing id is production state. Debug consumers get a new id or a non-committing assign.
• Commit the next offset, after the durable write, with commitSync on the poll thread.
• Leave auto-commit off. Add commitAsync only with a synchronous commit on revoke and on shutdown.
• Set CooperativeStickyAssignor explicitly on every member. Do not assume the default list selects it.
• Keep max.poll.interval.ms tight enough that a stuck put loses its partitions in minutes, and keep puts inside that bound.
• On CommitFailedException, stop the member. Do not continue as if you still own the partitions.
• Commit in onPartitionsRevoked. Do not commit in onPartitionsLost.
• Do not seek to earliest on assignment.
• Alert on per-partition lag for the landing group and, separately, on the Spark query. Reset one cursor only when that product must reprocess, with members stopped and a dry-run.
• Remember --to-datetime follows the broker timestamp, not payload event_time.

Common Mistakes and Anti-Patterns

Sharing the landing group id with a notebook or a console consumer. The notebook becomes a member, the assignment splits, and its commits move the lake's cursor. A new group id is cheaper than the rewind.

Committing the last processed offset without + 1 when you build the map yourself. The group rereads the last record forever. Use the no-argument commitSync after a full batch, or add one when you construct OffsetAndMetadata.

Auto-commit plus a slow sink. The timer records polled offsets before the put finishes, or fails to, depending on the race. Turn it off. The race is the bug.

Catching CommitFailedException and continuing. You no longer own the partitions. The new owner is consuming them. Your extra writes are either idempotent replays or duplicates. Crash the member and let the group converge.

Raising max.poll.interval.ms to hide rebalances. Rebalances are membership. The interval is a stuck-processing detector. Fix the assignor or the put. Do not disarm the detector.

Seeking to earliest in onPartitionsAssigned. Every deploy rewrites the retained log into the lake. Assignment is not a backfill request.

Resetting offsets while members are running. The next commit from a live member overwrites the reset, or the tool refuses. Stop the group. Dry-run. Execute. Start.

One reset for the tile and the landing group. Their cursors are different stores. Moving both to the same number skips or duplicates one of the products. Decide per product.

Renaming the group id as a versioning scheme. The new name has no commits. auto.offset.reset silently chooses a start. Version the application. Keep the group id stable unless a new cursor is the point, and then set the start explicitly.

Troubleshooting and Design Review

Symptom: Lag on the landing group grows on every partition during a deploy, then recovers. The assignor is the default. No put is slow.

Likely cause: eager rebalance revoking all partitions as each member restarts.

Investigation: member list over time, and partition.assignment.strategy in the running configuration. If the effective strategy is RangeAssignor, the default list did what it does.

Corrective action: set CooperativeStickyAssignor on all members and redeploy so that only the restarted member's partitions move. Confirm with describe during a restart drill. Do not widen the poll interval.

Symptom: CommitFailedException every few minutes. One member's put duration exceeds five minutes. Lag on its partitions oscillates.

Likely cause: max.poll.interval.ms exceeded. The member is ejected, commits, fails, retries the put, and gets ejected again.

Investigation: put latency next to the exception timestamps. Heartbeats may still be succeeding. That does not clear this symptom.

Corrective action: bound the put or lower max.poll.records so one poll's work finishes inside the interval. Restart the member cleanly. Idempotent objects should make the oscillation's rewrites safe. Verify one line per offset for the affected partitions.

Symptom: After a "small config change," the landing group reads from the tip and the raw prefix stops advancing even though produces continue. Lag is zero. The morning job is short.

Likely cause: the group id changed, or the commits were deleted, and auto.offset.reset=latest skipped the gap. Lag zero at the tip is consistent with a skip.

Investigation: the group id in the process arguments versus the catalog. The earliest raw offset you expected versus the current offset. A new group at latest has a current offset near the log end and no objects for the middle.

Corrective action: stop. If retention still holds the missing offsets, reset to the offset of the last good object and catch up. If not, record the gap. Restore the stable group id. Set the reset policy you meant, in the catalog, where the next config change can see it.

Symptom: Describe shows a member whose client.id is not a landing host. Partition assignment is uneven. Objects for some partitions stopped.

Likely cause: an ad-hoc consumer joined the group.

Investigation: --members --verbose. Identify the host. Check whether it committed.

Corrective action: stop the stranger. Compare committed offsets with the max offset in objects for the partitions it owned. Rewind those partitions if it committed past the last object. Leave partitions it did not touch. Leave the Spark checkpoint unless the tile was the stranger, which it should not be.

Symptom: --reset-offsets --to-datetime rebuilt the wrong hour. Payload event_time and broker timestamp differ by the upload delay.

Likely cause: the index is the broker timestamp type, not event_time.

Investigation: one raw line's event_time versus kafka_timestamp around the reset point.

Corrective action: choose offsets from the payload clock by scanning landed lines or a known-good manifest, then --to-offset. Re-run the replace. Note the timestamp type on the topic so the next rewind does not repeat the guess.

A design review asks: What is each group id, and who is forbidden to use it? What does the committed number mean? When does commitSync run? Which assignor is effective? What are the two timeouts? What happens on CommitFailedException? How is lag alerted, per partition? How do you rewind, and why is the Spark path not on that command? What does auto.offset.reset do the day the group already has commits? A review that answers "we commit offsets" has not reviewed offset management.

Interview Questions and Answers

Q1. What is a consumer group, and why do landing and the quality auditor not share one?

A group is one membership and one assignment of partitions, with one committed offset per partition. Members of the landing group split the twelve partitions so each record is landing's job once. The auditor needs every record as well, for a different product. A second group.id gets a second assignment of all partitions and a second commit stream. Putting the auditor in the landing group halves landing's partitions and moves the lake cursor when the auditor commits. Fan-out is multiple groups. Scaling a single job is multiple members of one group.

Q2. What does the committed offset mean, and where is it stored?

It is the next record the group should read, not the last record finished. After processing offsets 100 through 102, the commit is 103. commitSync() with no arguments writes that next position for the partitions in the last poll. The bytes live in the compacted internal topic __consumer_offsets, keyed by group and partition. They are not stored on the progress record. A restart fetches the committed next offset. If I build OffsetAndMetadata myself I add one to record.offset(). Forgetting the add replays the last record.

Q3. Why is auto-commit the wrong default for the landing job?

It commits the latest polled position on a timer inside poll, defaulting to five seconds. That moment is unrelated to the object-store acknowledgement. A commit that lands before the put becomes durable skips records after a crash. A commit that has not landed yet duplicates them. Which one you get is a race. I set enable.auto.commit=false and commitSync after the fsync and after the DLQ produce. The remaining crash duplicates a poll that the offset-named object absorbs.

Q4. How do session.timeout.ms and max.poll.interval.ms differ?

Session timeout is heartbeat loss. The background heartbeat thread can succeed while processing is stuck. max.poll.interval.ms is the maximum gap between poll calls, default five minutes. A put that blocks the poll thread longer than that ejects the member even though heartbeats are flowing. The following commitSync throws CommitFailedException because the generation has moved. I size puts to finish inside the interval, and I do not raise the interval to hide an eager rebalance.

Q5. What is the difference between an eager and a cooperative rebalance for this deploy?

Eager, which is what RangeAssignor does, revokes all partitions from every member and assigns from scratch. A rolling restart of landing stalls the whole topic once per restarted process. Cooperative sticky revokes only the partitions that change owners, so the other members keep fetching. I set CooperativeStickyAssignor explicitly, because a default preference list that starts with RangeAssignor will keep the group eager as long as every member supports range. Correctness of the offset-named object does not depend on this. Deploy lag does.

Q6. What do you commit in onPartitionsRevoked, and why not in onPartitionsLost?

In onPartitionsRevoked I still own the partitions, so I commitSync the work I have finished and the new owner starts after it. In onPartitionsLost the group has already taken them. A commit can fail or race the new member. I drop the local state and accept a replay from the last successful commit. The object key makes that replay a replace.

Q7. How do you compute lag, and why is per-partition lag the alert?

Lag is log end offset minus the committed next offset, which kafka-consumer-groups.sh --describe prints per partition. A sum across partitions can look small while partition 7 holds a day of data. The morning table is partitioned by the same logs. One stuck partition is a slice of missing raw data. I alert per partition on the landing group, and I alert the Spark query separately because its checkpoint does not appear in this describe output.

Q8. How do you rewind landing without destroying the tile?

I stop the landing members, dry-run --reset-offsets --to-offset for the partitions I mean, execute, and start the members. The idempotent writer replaces objects. I do not pass the Spark checkpoint to that command, and I do not delete it. If the tile's definition was also wrong, that is a second decision: a new or repaired checkpoint and a sink rewrite. --to-datetime follows the broker timestamp index. I map payload event_time to offsets myself before I reset.

Q9. What does auto.offset.reset do after the group has committed once?

Nothing, until the commits disappear or the group id changes. The setting applies only when there is no committed offset. earliest on a new landing group reads retained history. latest skips to the tip. Changing the property in a redeploy of an existing group does not move the cursor. A rename of group.id creates a new group and the property suddenly matters. I keep the group id stable and use --reset-offsets when I truly want a new position.

Q10. Why is the thirteenth landing member idle, and why is that not a reason to share the group with Spark?

There are twelve partitions, and a group assigns each partition to at most one member. A thirteenth member gets nothing until lesson 39 adds partitions, which changes the key mapping. Spark is not a fourteenth member of this group. It reads the same partitions through its own source and records progress in its checkpoint. Joining it to the landing group would make the micro-batch and the lake writer split partitions and commit one cursor for two side effects. I would rather add partitions on purpose than blur those cursors.

Hands-On Exercises

Exercise: Two members, a stranger group, and a dry-run rewind

Problem statement. Show that a group is one cursor, that a second group is a second cursor, and that a rewind is an explicit stopped-group operation.

Tasks.

1. Configure two landing members with CooperativeStickyAssignor, auto-commit off, and distinct client.id values. Subscribe with a revoke listener that commits.
2. Demonstrate assignment before and after the second member joins. Publish records and show one object per offset.
3. Run a reader with a different group.id and show that landing's committed offsets do not follow it.
4. Force a poll gap longer than max.poll.interval.ms in a lab hook, capture CommitFailedException, and show that a clean restart continues from the last good commit.
5. Stop the members. Dry-run --reset-offsets --to-offset for one partition. Execute only if you can replace objects. Show the replaced object still has one copy of each offset.
6. State the Spark checkpoint path and confirm your command did not take it as an argument.
7. Explain why --to-datetime is the wrong tool for "replay event_time after 18:00 UTC."

Expected deliverables. Describe transcripts, the exception line, the dry-run output, and the checkpoint paragraph.

Evaluation rubric.

You pass when a reviewer can see two group ids with different current offsets on the same partition, and a rewind plan that does not mention the checkpoint.

Lesson Summary

learning-progress-raw-landing is one assignment of the twelve progress partitions and one committed next-offset per partition, stored in __consumer_offsets. The quality auditor is a second group. The activity tile is a checkpoint, not a member of either group. commitSync runs after the durable write, on the poll thread, including in onPartitionsRevoked, and not in onPartitionsLost. Auto-commit stays off because its timer is not the sink. The cooperative sticky assignor is set explicitly so a deploy does not revoke every partition. max.poll.interval.ms bounds processing. Exceeding it yields CommitFailedException and a member that must stop. Lag is per partition, log end minus committed next offset. A rewind stops the group, dry-runs --to-offset, and leaves the Spark checkpoint to its own runbook. auto.offset.reset matters when the group has no commit, including the day someone renames the group id.

Kafka Delivery Semantics and Transactions names what this commit pattern guarantees, what the idempotent producer already covered, and what a transaction does and does not include. The primer on the protocol is Rebalancing and Offset Management.

Further Reading

• Consumer configurations — group.id, enable.auto.commit, max.poll.interval.ms, session.timeout.ms, partition.assignment.strategy, group.instance.id.
• KafkaConsumer Javadoc — committed offsets as the next record, commitSync, wakeup, and rebalance listener behavior.
• kafka-consumer-groups.sh — describe, members, and --reset-offsets. Confirm flags against the distribution you run.
• Consumer Groups, Manual Commit, and Rebalance Issues — the specialty primer.
• Structured Streaming with Spark — the checkpoint this lesson refuses to reset by group command.