Commit the Completed Kafka Prefix, Not the Last Completed Record

Three handlers start for offsets 10, 11, and 12. Ten finishes. Twelve finishes. Eleven is still waiting on a dependency when the process prepares to commit.

Committing 13 looks efficient. It is also wrong. Normal group recovery starts at the stored next offset, so a replacement starts after the unfinished record at 11.

The safe boundary is 11 until that handler finishes. This article builds the application ledger needed to compute that boundary, then adds sparse offsets, rebalance, commit callbacks, shutdown, and poison data. The reference profile is Apache Kafka and the Java client 4.3.1 with traditional consumer groups. Share groups use a different acquisition model and appear in their own reference.

Kafka exposes positions; your application owns completion

Four positions often collapse into one variable:

The Kafka 4.3 consumer API distinguishes position from committed offsets. ConsumerRecords.nextOffsets() returns next-offset metadata for partitions in a poll. Use that metadata when available instead of deriving progress from record count.

None of these positions proves that an external API call or database mutation completed. A consumer can have correct Kafka progress and still repeat an unprotected effect after a crash.

The first unfinished delivered record is the commit ceiling

Track every delivered record that can still block progress, including records from earlier polls. Each entry needs:

• partition and offset;
• pending, completed, or durably quarantined state;
• the ownership incarnation that dispatched it;
• the client-supplied next-offset/leader-epoch metadata where available.

The basic frontier is short:

interface DeliveredWork {
  readonly offset: number
  readonly state: "pending" | "completed" | "quarantined"
  readonly ownershipToken: string
}
 
const completedFrontier = (
  delivered: readonly DeliveredWork[],
  fetchedNext: number,
  ownershipToken: string,
): number => {
  const firstUnfinished = delivered
    .filter((work) => work.ownershipToken === ownershipToken)
    .toSorted((left, right) => left.offset - right.offset)
    .find((work) => work.state === "pending")
 
  return firstUnfinished?.offset ?? fetchedNext
}

The short function hides important preconditions. delivered must contain all unfinished work, not only the latest poll. quarantined means a required durable disposition exists, not that an error was logged. A completion from an old ownership token cannot mutate the new ledger.

With 10 and 12 complete and 11 pending, the frontier is 11. Once 11 completes, the sorted ledger has no pending entries and the frontier advances through 12 to fetched next offset 13.

Compare C02 with C03. The unsafe branch commits the last completed offset plus one, crashes, then resumes from 13. Kafka did not delete record 11; the application stored progress past it.

Sparse offsets must come from source evidence, not guessed holes

Kafka offsets are positions, not a visible-record counter. A returned batch can contain offsets 10 and 12 without delivering 11 because a control record, aborted transaction, or compacted record occupies or removed that position. The message-format reference and KafkaConsumer isolation behavior make those gaps legitimate.

If the actual delivered ledger contains 10 and 12, completing 10 can advance the frontier to 12. Do not invent a pending handler for 11. If the application really dispatched 11, however, it belongs in the ledger and caps the frontier.

This rule is why “start offset + number of records” is unsafe. It loses the client's real next-offset boundary and any sparse history.

The consumer-owning thread should own progress decisions

KafkaConsumer is not thread-safe. Background handlers can process records, but they should return bounded completion messages to the thread that owns polling, pause/resume, seek, and commit decisions. The API names wakeup() as the special cross-thread operation.

The following Java is an annotated design sketch, not a production SDK:

while (running) {
    ConsumerRecords<String, byte[]> records =
        consumer.poll(Duration.ofMillis(250));                    // ①
 
    ledger.registerReturned(records, records.nextOffsets());      // ②
    for (ConsumerRecord<String, byte[]> record : records) {
        if (!admission.tryReserve(record.serializedValueSize())) {
            consumer.pause(Set.of(
                new TopicPartition(record.topic(), record.partition())));
            pendingDispatch.add(record);                          // ③
            continue;
        }
        workers.submit(() ->
            completions.put(process(record, ownershipToken, record)));
    }
 
    dispatchPendingWithinCapacity(pendingDispatch, workers, completions);
 
    Completion completion;
    while ((completion = completions.poll()) != null) {
        ledger.applyIfCurrent(completion);
    }
 
    Map<TopicPartition, OffsetAndMetadata> next =
        ledger.completedFrontiers(records.nextOffsets());
    consumer.commitSync(next);
}

① Poll stays on the owner thread. max.poll.records limits returned records, not all prefetched bytes; the application still needs record and byte budgets.

② Every returned record enters the pending ledger before dispatch. The partition's nextOffsets() value can cap progress only after every earlier returned record reaches the selected completion/disposition state.

③ Capacity pressure pauses future delivery and retains already returned work in a bounded local queue. Pausing cannot put a record back into the prior ConsumerRecords batch. Workers still receive an ownership token, and a delayed completion from a revoked/lost incarnation is discarded locally. That does not cancel an already dispatched external request.

The sketch omits the concrete bounded-queue implementation, executor rejection, error classification, rebalance callbacks, and shutdown coordination. If the pending queue is full, stop dispatch/poll admission without advancing the frontier. Use the sketch to review ownership, not as a copy-and-paste wrapper.

Pause controls admission; it is not cancellation

When one assigned partition reaches the in-flight cap, pause it and keep the required poll lifecycle moving for other assigned partitions. pause() does not:

• cancel handlers already running;
• reclaim every byte the client prefetched;
• extend business deadlines;
• preserve ownership through a future reassignment.

The consumer configuration reference also permits an oversized first record batch beyond a per-partition fetch limit so the client can make progress. Set a payload-size contract as well as a record-count contract.

Revocation, loss, and downstream fencing are separate events

A normal revocation callback can provide a bounded cleanup opportunity. An assignment may also be lost, meaning another member can already own it. ConsumerRebalanceListener separates those callbacks.

In both cases, an old process can still finish a dispatched HTTP request. Kafka can reject stale group progress when the request carries stale metadata; only the downstream resource can reject a stale effect. Use its operation key, conditional write, fencing token, or reconciliation API.

Static membership changes reassignment timing. KIP-848 moves more heartbeat, session, and assignment policy to the server. Neither feature places a fence inside your database or provider. The KIP-848 migration article owns that protocol rollout.

Java async commit ordering does not excuse an old new request

Java orders commitAsync invocations and their callbacks. The client does not arbitrarily deliver an older callback after a newer callback and then retry the old request behind your back. The source path through ConsumerCoordinator preserves invocation ordering.

An application can still cause regression:

1. request snapshot 13;
2. observe it stored;
3. later issue a new commit invocation carrying stale snapshot 11.

The coordinator can store 11. That is an application snapshot bug, not proof that Java reordered callbacks. Keep one owner for commit sequencing and discard snapshots from older processing/ownership epochs.

Poison data needs a named failure stage

“Deserialization failed” can mean two different paths:

1. The Kafka client's configured deserializer throws before a normal handler receives a record.
2. A raw-byte consumer receives source coordinates, then an application decoder rejects the bytes.

The second path can build a quarantine record with original coordinates and synthetic raw evidence. The first needs a verified client/framework-specific recovery method; do not pretend the handler has a usable record when it never received one.

In either path, progress advances only after an explicit disposition contract. Plain KafkaConsumer does not create a universal DLQ. A same-cluster Kafka transaction can join a quarantine record and source offsets, as the recovery guide demonstrates.

Shutdown preserves only completed work inside remaining authority

A bounded shutdown sequence is:

1. stop admitting new work;
2. keep the required consumer lifecycle alive while ownership remains valid;
3. drain until the earlier of completion, grace expiry, or ownership loss;
4. commit only the completed owned frontier;
5. close.

Forced exit does not roll back a request that already reached an external provider. If the grace expires, label the rest unresolved and let normal recovery plus the sink identity contract decide what can repeat.

What could go wrong

• Only the latest poll stays in the ledger. A pending record from an earlier poll disappears from the frontier calculation.
• A sparse offset is treated as unfinished work. The consumer waits forever on a handler that never existed.
• Auto commit tracks fetched progress while workers are still running. A crash resumes after incomplete work.
• Every revoke is treated as graceful. onPartitionsLost does not grant a final safe commit window.
• Pause is treated as a memory ceiling. Prefetched bytes and oversized batches can still exceed the record-count intuition.
• A successful Kafka commit is treated as effect atomicity. Redelivery can repeat an external effect even with perfect frontier logic.

The offset you may commit is the first position the replacement can start from without skipping unfinished owned work. Everything after that word “owned” needs another effect contract.

Sources and references

• KafkaConsumer 4.3.1 source
• Kafka 4.3 consumer configuration
• ConsumerRebalanceListener 4.3.1
• ConsumerCoordinator 4.3.1
• Kafka 4.3 message format
• Kafka 4.3 Consumer Rebalance Protocol