Apache Kafka — Module 12Intermediate+150 XP

Consumer Design

How to build a production-grade Kafka consumer: the full config walkthrough, manual commit-after-processing, idempotent processing, poison-message handling, graceful shutdown, rebalance listeners, and a production-readiness checklist.

60 min September 2026

Producer Design Kafka Connect

// Part 01 — From concept to client

A Consumer That Just Reads Records Is the Easy 10%

Module 03 established the mechanics: consumers pull records by calling poll(), track progress through offsets, and belong to consumer groups that share partition ownership. That module also flagged, without fully resolving, the hardest part of consumer design — the separation between processing a record and committing its offset, and everything that can go wrong in the gap between those two actions.

This module closes that gap with real, applied code. As in the producer module, we use Python with the confluent-kafka client throughout, consistently, so the property names and control flow map directly onto whatever client library you actually use in production. A production consumer is not a poll() loop with a process() call inside it — it is a poll loop plus deliberate answers to five questions: when do I commit, how do I survive reprocessing the same record twice, what happens to a record that can never be processed successfully, how do I shut down without losing my place, and what do I do the instant a partition is about to be taken away from me.

What "production-ready" means for a consumer, concretely: a crash between processing and committing never silently skips a record. Reprocessing the same record twice — which at-least-once delivery guarantees will eventually happen — never corrupts downstream state. A single malformed record does not permanently stall the partition it lives on. A rebalance does not lose or duplicate in-flight work. And shutdown commits the true, current position before the process exits, not whatever was committed five seconds ago on the auto-commit timer.

Part 06 builds toward a complete OrderEventConsumer class that ties every one of these concerns together. Everything before it is groundwork for why that class is shaped the way it is.

// Part 02 — The configuration surface

Every Property That Actually Matters, and What Its Default Costs You

Consumer configuration has the same shape as producer configuration from the previous module: a handful of properties actually determine reliability, and several defaults are tuned for convenience rather than production safety.

group.id — the identity that makes everything else work

Every consumer must belong to a consumer group, identified by group.id. This is not optional metadata — it is the key the broker uses to track committed offsets (stored in the internal __consumer_offsets topic, per Module 03) and to determine which consumers share partition ownership. Two processes with the same group.id reading the same topic split the partitions between them. Two processes with different group.id values each get an independent, full copy of every partition. Getting this wrong — accidentally sharing agroup.id between two unrelated services, or accidentally giving every instance of the same service a unique one — is one of the most common consumer bugs, and it does not throw an error; it just silently produces the wrong fan-out behavior.

auto.offset.reset — what happens with no committed offset

When a consumer group has no committed offset for a partition — a brand-new group, or a partition whose committed offset has aged out of the __consumer_offsets topic's own retention —auto.offset.reset decides where to start. earliest starts from the oldest retained record; latest (a common client default) starts from the next record produced after the consumer connects, skipping everything already in the topic. For a new service backfilling historical state, earliest is almost always correct. For a service that only cares about events going forward — a live notification service, for instance — latest avoids an enormous, unwanted backlog on first startup. This setting only takes effect when there is no valid committed offset; it does not override a real committed position.

enable.auto.commit — the setting most responsible for silent data loss

As Module 03 covered, enable.auto.commit=true (the default in most client libraries) commits the latest offset returned by poll() on a fixed timer (auto.commit.interval.ms, typically 5 seconds), independent of whether your application has actually finished processing those records. This module treats enable.auto.commit=falsecombined with explicit, manual commits after real processing success as the default posture for any consumer whose work has a real side effect — writing to a database, calling a payment API, sending a notification. Part 03 builds the manual-commit pattern in full.

max.poll.records and max.poll.interval.ms — the processing-time budget

max.poll.records caps how many records one poll() call returns.max.poll.interval.ms is the maximum time allowed between successive poll()calls before the group coordinator presumes the consumer is stuck and triggers a rebalance. These two settings must be sized together: if processing max.poll.records worth of records at your actual per-record processing time can exceed max.poll.interval.ms, you will see rebalances under load that have nothing to do with an actual crash.

session.timeout.ms — detecting a genuinely dead consumer

Separately from the poll-interval mechanism, a background heartbeat thread (in most client libraries) pings the group coordinator on heartbeat.interval.ms. If no heartbeat arrives within session.timeout.ms, the coordinator presumes the process itself is dead — crashed, network-partitioned, or frozen — and triggers a rebalance independent of whatever the main thread's poll() timing looks like.

isolation.level — visibility into transactional writes

If the topics this consumer reads from are written by a transactional producer (covered in the message brokers module's exactly-once section), isolation.level=read_committed ensures the consumer only sees records from transactions that actually committed, filtering out records from transactions that were aborted or are still in flight. The default, read_uncommitted, exposes every write regardless of transaction outcome — usually the wrong choice whenever transactional producers are anywhere in the pipeline.

Property	What it controls	Production-safe default
group.id	Identity used for offset tracking and partition-sharing between consumers.	One stable value per logical service; never shared across unrelated services.
auto.offset.reset	Where to start reading when no valid committed offset exists.	'earliest' for services needing full history; 'latest' for forward-only event consumers.
enable.auto.commit	Whether offsets commit on a timer, independent of processing outcome.	false for any workflow with real side effects — pair with explicit manual commits.
max.poll.records	Max records returned per poll() call.	Sized so max.poll.records × real per-record processing time stays well under max.poll.interval.ms.
max.poll.interval.ms	Max time allowed between poll() calls before a rebalance is triggered.	300000 (5 min) default; raise it, or lower max.poll.records, to match real processing time.
session.timeout.ms	Max time without a heartbeat before the consumer is presumed dead.	10-45 seconds depending on client version; tune alongside GC pause expectations for JVM clients.
isolation.level	Whether uncommitted transactional writes are visible.	'read_committed' whenever upstream producers use transactions.

These settings interact — never tune one in isolation

max.poll.records and max.poll.interval.ms are a pair, not independent knobs — Module 03's rebalance timeline shows exactly how tuning one without the other produces a mysterious rebalance under load. The same discipline applies here: change the ratio deliberately, and re-verify under realistic per-record processing time, not synthetic benchmarks with a no-op process function.

// Part 03 — Manual commit-after-processing

Commit Only After Work Is Actually, Durably Done

The core reliability pattern for a consumer with real side effects is simple to state and easy to get subtly wrong in practice: process a record completely — including any database write, API call, or downstream produce — before committing its offset, and never commit an offset for work that has not fully and durably succeeded.

manual commit after processing — the baseline pattern

from confluent_kafka import Consumer, KafkaError

consumer = Consumer({
    'bootstrap.servers': 'broker-1:9092,broker-2:9092',
    'group.id': 'order-processing-service',
    'auto.offset.reset': 'earliest',
    'enable.auto.commit': False,   # we commit manually, only after success
    'max.poll.records': 200,
    'max.poll.interval.ms': 300000,
})
consumer.subscribe(['orders.events'])

try:
    while True:
        msg = consumer.poll(timeout=1.0)
        if msg is None:
            continue
        if msg.error():
            if msg.error().code() == KafkaError._PARTITION_EOF:
                continue
            logger.error(f"Consumer error: {msg.error()}")
            continue

        try:
            process_order_event(msg)          # the real work — must be durable
            consumer.commit(message=msg, asynchronous=False)
            # commit only happens AFTER process_order_event fully succeeds
        except Exception as exc:
            logger.error(f"Processing failed, offset NOT committed: {exc}")
            # do not commit -- this record will be redelivered on restart
            raise
finally:
    consumer.close()

Committing after every single record, as shown above, is the safest pattern but also the slowest — each commit(asynchronous=False) call is itself a synchronous write to the__consumer_offsets topic, and doing this per record adds real latency under high throughput. In practice, most production consumers commit after each successfully processedbatch returned by one poll() call, not after each individual record, trading a slightly larger reprocessing window on crash for meaningfully better throughput.

batch-level manual commit — the common production pattern

while True:
    records = consumer.poll(timeout=1.0)
    batch = consumer.consume(num_messages=200, timeout=1.0)
    if not batch:
        continue

    try:
        for msg in batch:
            if msg.error():
                continue
            process_order_event(msg)   # every record in the batch must succeed

        # commit once, for the whole batch, only after all records succeeded
        consumer.commit(asynchronous=False)
    except Exception as exc:
        logger.error(f"Batch processing failed, offsets NOT committed: {exc}")
        # entire batch will be redelivered from the last committed offset
        raise

Commit granularity	Reprocessing window on crash	Throughput cost
Per record	At most one record reprocessed.	Highest — one offset-topic write per record.
Per batch (per poll() call)	Up to one full batch reprocessed.	Low — one offset-topic write per batch, amortized across many records.
Fixed time interval (e.g. every 5s)	Everything processed since the last interval reprocessed.	Lowest, but reintroduces some of auto-commit's timing risk if not tied to actual processing completion.

Batch commits only work if every record in the batch is genuinely idempotent to reprocess

A per-batch commit means a crash mid-batch redelivers the entire batch, including records that were already fully processed before the crash. This is not a flaw to work around — it is the at-least-once contract Kafka gives you by design. Part 04 covers the processing-side pattern that makes this safe.

// Part 04 — Idempotent processing

Reprocessing Will Happen. Design So It Never Corrupts State.

At-least-once delivery is not a bug to be engineered away — it is the fundamental guarantee a commit-after-processing consumer provides, and any consumer that commits after doing real work will occasionally reprocess a record it already handled. The only durable fix is making the processing logic itself idempotent: applying the same record twice produces the same end state as applying it once.

The wrong instinct: trying to prevent redelivery

Engineers new to this problem often try to solve it by committing more aggressively, or by adding complex distributed locking to prevent a record from ever being processed twice. Both approaches fight the delivery model instead of accepting it. Kafka's own transactional exactly-once guarantees (covered in the message brokers module) only apply within a Kafka-to-Kafka pipeline — the moment your consumer's side effect is a database write, an external API call, or a notification send, you are back to at-least-once semantics at that boundary, and idempotent processing is the only reliable answer.

Pattern 1 — idempotency keys on the write itself

The most robust pattern: every event carries a stable, unique identifier, and every downstream write checks or enforces uniqueness on that identifier before applying the write's effect.

idempotent processing via a unique constraint at the database level

def process_order_event(msg):
    event = json.loads(msg.value())
    order_id = event['order_id']
    event_id = event['event_id']  # unique per logical event, assigned at production time

    # rely on a unique constraint on event_id at the database level --
    # a duplicate insert raises an integrity error instead of double-applying
    try:
        db.execute(
            "INSERT INTO order_events (event_id, order_id, status, processed_at) "
            "VALUES (%s, %s, %s, now())",
            (event_id, order_id, event['status']),
        )
    except UniqueConstraintViolation:
        logger.info(f"event_id={event_id} already processed, skipping duplicate")
        return  # not an error -- this is exactly-once processing working as intended

Pattern 2 — idempotent by construction, using upserts

When the event represents a state transition rather than an append-only fact, an upsert keyed on the entity ID is naturally idempotent — applying the same "order status = shipped" event twice leaves the row in exactly the same state either time, with no special-case duplicate detection required at all.

idempotent by construction — upsert instead of append

def process_order_status_event(msg):
    event = json.loads(msg.value())

    # UPSERT: applying this twice with the same data is a no-op the second time.
    # No explicit duplicate check needed -- idempotency is structural.
    db.execute(
        "INSERT INTO order_status (order_id, status, updated_at) "
        "VALUES (%s, %s, %s) "
        "ON CONFLICT (order_id) DO UPDATE SET "
        "  status = EXCLUDED.status, updated_at = EXCLUDED.updated_at "
        "  WHERE EXCLUDED.updated_at > order_status.updated_at",
        (event['order_id'], event['status'], event['event_timestamp']),
    )
    # the WHERE clause also protects against OUT-OF-ORDER redelivery --
    # an older event replayed after a newer one is a no-op, not a regression

Pattern	When to use it	What it protects against
Unique constraint on event_id	Append-only event logs (audit trails, event sourcing).	Exact duplicate reprocessing of the same event.
Upsert keyed on entity ID	State-representing events (current status, current balance).	Duplicate AND out-of-order reprocessing, if timestamps are compared in the upsert.
Idempotency key on an external API call	Side effects outside your own database (payment charges, sending an email).	Duplicate external side effects, which are often far more costly than a duplicate database row.

idempotent external API calls using a client-supplied idempotency key

def process_payment_event(msg):
    event = json.loads(msg.value())

    # many payment APIs (Stripe among them) accept a client-supplied
    # idempotency key; the same key submitted twice returns the original
    # result instead of charging twice
    payment_client.create_charge(
        amount=event['amount_cents'],
        customer_id=event['customer_id'],
        idempotency_key=event['event_id'],  # stable across redeliveries
    )

Idempotency keys must come from the event, never be generated at processing time

Generating a fresh idempotency key each time a record is processed defeats the entire purpose — the key must be assigned once, at production time, and carried through the event so that every redelivery of the same logical event carries the same key. A key derived fromdatetime.now() or a random UUID generated inside the consumer is not an idempotency key at all.

// Part 05 — Poison messages and the DLQ

One Bad Record Should Never Block Every Record Behind It

A poison message is a record that fails processing every time it is attempted — a malformed payload, a schema violation, a value the processing logic cannot handle regardless of how many times it retries. Because a consumer following Part 03's pattern only advances its committed offset after successful processing, a poison message that is retried forever blocks the entire partition: nothing after it can be processed until it either succeeds or is explicitly skipped.

Retry with a bounded count, then dead-letter

The pattern from the message brokers module's dead letter queue coverage applies directly here: retry a failing record a small, bounded number of times to absorb genuinely transient failures, then route it to a DLQ topic and commit past it, rather than retrying indefinitely or crashing the whole consumer.

poison message handling — bounded retry, then route to DLQ

import json
import logging

logger = logging.getLogger(__name__)
MAX_RETRIES = 3

def process_with_dlq(msg, process_fn, dlq_producer, dlq_topic):
    """
    Attempt processing up to MAX_RETRIES times. On exhaustion, write the
    failed record to the DLQ and return normally so the caller commits
    past it -- this is what actually unblocks the partition.
    """
    last_exception = None
    event = json.loads(msg.value())

    for attempt in range(1, MAX_RETRIES + 1):
        try:
            process_fn(event)
            return True  # success -- caller commits normally
        except Exception as exc:
            last_exception = exc
            logger.warning(
                f"Processing failed (attempt {attempt}/{MAX_RETRIES}): "
                f"event_id={event.get('event_id')} error={exc}"
            )

    # retries exhausted -- route to DLQ instead of blocking the partition forever
    dlq_event = {
        'original_event': event,
        'error_message': str(last_exception),
        'error_type': type(last_exception).__name__,
        'retry_count': MAX_RETRIES,
        'source_topic': msg.topic(),
        'source_partition': msg.partition(),
        'source_offset': msg.offset(),
    }
    dlq_producer.produce(
        topic=dlq_topic,
        key=msg.key(),
        value=json.dumps(dlq_event).encode('utf-8'),
    )
    dlq_producer.flush()
    logger.error(f"Event sent to DLQ: event_id={event.get('event_id')}")
    return True  # caller commits -- this record is "handled," just not successfully

The critical design decision is the return value: process_with_dlq returnsTrue in both the success case and the exhausted-retries-routed-to-DLQ case. Both are "handled" from the offset-commit perspective — the difference between them is tracked in the DLQ and in metrics, not in whether the offset advances. Only a genuinely unexpected exception (a bug in the DLQ write itself, for instance) should prevent the commit and stop the consumer loop.

Distinguishing transient failures from genuinely poison ones

Not every failure on the first attempt is a poison message — a downstream database being briefly unavailable looks identical, on attempt one, to a record that will never succeed. The bounded retry count exists precisely to absorb the transient case without over-engineering a distinction that a simple retry-then-dead-letter policy already handles well in practice. Teams that need finer distinction typically add a short, exponential backoff between attempts rather than immediate retries, so a transient failure has time to actually resolve before the retry budget is exhausted.

adding backoff between retries

import time

for attempt in range(1, MAX_RETRIES + 1):
    try:
        process_fn(event)
        return True
    except Exception as exc:
        last_exception = exc
        if attempt < MAX_RETRIES:
            backoff_seconds = 2 ** attempt  # 2s, 4s, 8s
            logger.warning(f"Retry {attempt} failed, backing off {backoff_seconds}s")
            time.sleep(backoff_seconds)

A sleeping consumer thread is not calling poll() — watch max.poll.interval.ms

Backoff sleeps happen inside the processing path, between poll() calls, which means they count against your max.poll.interval.ms budget just like slow processing does. A retry policy with long backoffs multiplied by a large max.poll.records can trip a rebalance for the same reason covered in Part 02 — size the two together.

Reading record headers before committing to a deserialization path

Mirroring the producer side's use of headers, a consumer can inspect a record's headers cheaply before deciding how to deserialize its value — useful during a schema migration when a topic temporarily carries two payload versions, or when routing certain records to different handling logic based on metadata alone.

using headers to route before committing to full deserialization

def process_order_event(msg):
    headers = dict(msg.headers() or [])
    schema_version = headers.get('schema_version', b'1').decode()

    if schema_version == '1':
        event = parse_legacy_order_event(msg.value())
    else:
        event = parse_order_event_v3(msg.value())

    # trace_id from headers lets this consumer's processing span
    # connect back to the producing service's original trace, without
    # the trace ID needing to live inside the business payload itself
    trace_id = headers.get('trace_id', b'').decode()
    with tracer.start_span('process_order_event', trace_id=trace_id):
        process_with_dlq(msg, event)

This pattern is also what makes a poison-message DLQ record genuinely useful for debugging: if the original producer attached a producing_service and trace_id header, the DLQ event in Part 05 can carry that context forward automatically, connecting a failed record all the way back to the request that originally produced it.

// Part 06 — The complete worked example

A Production-Grade Order-Events Consumer, Start to Finish

This class combines manual batch commits, idempotent processing, bounded-retry DLQ routing, and a rebalance listener into one consumer you could adapt directly for a real order-processing service.

order_event_consumer.py — configuration and initialization

import json
import logging
import signal
import sys

from confluent_kafka import Consumer, Producer, KafkaError, TopicPartition

logger = logging.getLogger(__name__)
MAX_RETRIES = 3


class OrderEventConsumer:
    """
    Production consumer for order lifecycle events.
    Reliability posture: manual commit after successful batch processing,
    idempotent database writes, bounded retry with DLQ fallback for
    poison messages, and offset commits on partition revocation so a
    rebalance never silently reprocesses more than necessary.
    """

    def __init__(self, bootstrap_servers: str, topic: str, dlq_topic: str, group_id: str):
        self.topic = topic
        self.dlq_topic = dlq_topic
        self._running = True

        self.consumer = Consumer({
            'bootstrap.servers': bootstrap_servers,
            'group.id': group_id,
            'auto.offset.reset': 'earliest',
            'enable.auto.commit': False,
            'max.poll.records': 200,
            'max.poll.interval.ms': 300000,
            'session.timeout.ms': 20000,
            'isolation.level': 'read_committed',
        })

        self.dlq_producer = Producer({
            'bootstrap.servers': bootstrap_servers,
            'acks': 'all',
            'enable.idempotence': True,
        })

        self.consumer.subscribe(
            [topic],
            on_assign=self._on_partitions_assigned,
            on_revoke=self._on_partitions_revoked,
        )

order_event_consumer.py — the rebalance listener

    def _on_partitions_assigned(self, consumer, partitions):
        logger.info(f"Partitions assigned: {[p.partition for p in partitions]}")
        # nothing to do here in the common case -- the consumer will
        # simply resume from the last committed offset for each partition

    def _on_partitions_revoked(self, consumer, partitions):
        """
        Called BEFORE partitions are taken away, whether from a rebalance
        or a clean shutdown. This is the last chance to commit work that
        has already been processed but not yet committed -- skipping this
        means that work gets reprocessed by whichever consumer picks up
        the partition next, even though it already succeeded here.
        """
        logger.info(f"Partitions revoked: {[p.partition for p in partitions]}, committing")
        try:
            consumer.commit(asynchronous=False)
        except Exception as exc:
            # a failed commit here is not fatal to the rebalance, but it
            # does mean the next owner will reprocess more than necessary
            logger.error(f"Commit during partition revocation failed: {exc}")

order_event_consumer.py — idempotent processing logic

    def _process_order_event(self, event: dict):
        order_id = event['order_id']
        event_id = event['event_id']

        # upsert keyed on order_id with a newer-timestamp guard --
        # idempotent AND safe against out-of-order redelivery
        db.execute(
            "INSERT INTO order_status (order_id, event_id, status, updated_at) "
            "VALUES (%s, %s, %s, %s) "
            "ON CONFLICT (order_id) DO UPDATE SET "
            "  status = EXCLUDED.status, event_id = EXCLUDED.event_id, "
            "  updated_at = EXCLUDED.updated_at "
            "  WHERE EXCLUDED.updated_at > order_status.updated_at",
            (order_id, event_id, event['status'], event['event_timestamp']),
        )

order_event_consumer.py — bounded retry with DLQ fallback

    def _process_with_dlq(self, msg) -> bool:
        event = json.loads(msg.value())
        last_exception = None

        for attempt in range(1, MAX_RETRIES + 1):
            try:
                self._process_order_event(event)
                return True
            except Exception as exc:
                last_exception = exc
                logger.warning(
                    f"Processing failed (attempt {attempt}/{MAX_RETRIES}): "
                    f"event_id={event.get('event_id')} error={exc}"
                )

        dlq_event = {
            'original_event': event,
            'error_message': str(last_exception),
            'error_type': type(last_exception).__name__,
            'retry_count': MAX_RETRIES,
            'source_partition': msg.partition(),
            'source_offset': msg.offset(),
        }
        self.dlq_producer.produce(
            topic=self.dlq_topic,
            key=msg.key(),
            value=json.dumps(dlq_event).encode('utf-8'),
        )
        self.dlq_producer.flush()
        logger.error(f"Routed to DLQ after {MAX_RETRIES} attempts: event_id={event.get('event_id')}")
        return True  # handled -- offset still advances past this record

order_event_consumer.py — the main loop and graceful shutdown

    def run(self):
        signal.signal(signal.SIGTERM, self._handle_shutdown_signal)
        signal.signal(signal.SIGINT, self._handle_shutdown_signal)

        try:
            while self._running:
                msg = self.consumer.poll(timeout=1.0)
                if msg is None:
                    continue
                if msg.error():
                    logger.error(f"Consumer error: {msg.error()}")
                    continue

                self._process_with_dlq(msg)
                self.consumer.commit(message=msg, asynchronous=False)
        finally:
            self.close()

    def _handle_shutdown_signal(self, signum, frame):
        logger.info(f"Received signal {signum}, requesting shutdown")
        self._running = False
        # consumer.wakeup() interrupts a blocking poll() call immediately,
        # rather than waiting for its timeout to elapse naturally
        self.consumer.wakeup()

    def close(self):
        logger.info("Shutting down consumer, committing final offsets")
        try:
            self.consumer.commit(asynchronous=False)
        except Exception as exc:
            logger.error(f"Final commit during shutdown failed: {exc}")
        self.dlq_producer.flush(timeout=10)
        self.consumer.close()
        logger.info("Consumer shutdown complete")

Every piece traces back to a named concern from earlier parts: enable.auto.commit=Falseplus explicit commit() calls from Part 03, the upsert-with-timestamp-guard idempotent write from Part 04, the bounded-retry-then-DLQ pattern from Part 05, and — new here —on_assign/on_revoke callbacks that commit before a partition is taken away, plus consumer.wakeup() wired to signal handling so shutdown interrupts a blockingpoll() immediately instead of waiting out its timeout.

// Part 07 — Graceful shutdown in depth

consumer.wakeup() and Why a Blocking poll() Needs Interrupting

consumer.poll(timeout=1.0) blocks for up to one second waiting for records. During normal operation this is harmless — one second of added shutdown latency is nothing. But a consumer in the middle of processing a slow record when a shutdown signal arrives needs a way to interrupt that wait cleanly rather than relying on the next timeout to happen to notice a shutdown flag.consumer.wakeup(), called from a signal handler, interrupts a blocking poll()call immediately by raising an exception inside it, letting the main loop's exit condition be checked right away instead of up to a full timeout seconds later.

the full shutdown sequence, in order

# 1. SIGTERM arrives from the orchestrator
# 2. signal handler sets self._running = False
# 3. signal handler calls consumer.wakeup() -- interrupts any blocking poll() now
# 4. main loop's next poll() call raises, or the loop's while condition
#    is checked and exits cleanly
# 5. finally block calls close()
# 6. close() commits the current position with a final synchronous commit
# 7. close() flushes the DLQ producer (in case a DLQ write was still in flight)
# 8. close() calls consumer.close(), which also triggers a final
#    partition-revocation cycle and a LeaveGroup request to the coordinator,
#    letting the group rebalance immediately rather than waiting out
#    session.timeout.ms for a departure the coordinator doesn't yet know about

That last point matters more than it looks: an unclean shutdown — a killed process that never callsconsumer.close() — leaves the group coordinator waiting out the fullsession.timeout.ms before it notices the consumer is gone and reassigns its partitions. A clean shutdown that calls close() sends an explicit departure notice, and the group rebalances immediately. For a rolling deploy restarting many consumer instances in sequence, this difference is the gap between a smooth deploy and every deploy triggering asession.timeout.ms-long stall on each instance.

Match the flush and commit timeouts to your orchestrator's grace period

Exactly as with the producer module's shutdown guidance, the final commit and DLQ flush inside close() need to complete before the orchestrator's SIGKILL arrives. Keep this sequence fast and bounded — it should not itself be doing slow, unbounded work during shutdown.

// Part 08 — Rebalance listeners in depth

onPartitionsRevoked Is Your Last Chance Before the Work Is Gone

A rebalance listener's on_revoke callback (the naming varies by client library — Java's consumer API calls this onPartitionsRevoked) fires before the group coordinator actually reassigns a consumer's partitions to someone else. This is the precise moment to commit any processed-but-not-yet-committed offsets for the partitions about to be taken away — after this callback returns, the new owner of those partitions resumes from whatever was last committed, with no further chance for the outgoing consumer to influence that starting point.

What happens if you skip the revoke callback

Without committing in on_revoke, a consumer relying solely on its normal per-batch commit cadence can lose work at exactly the wrong moment: a rebalance triggered mid-batch means the current batch's successfully processed records are never committed, because the partition is reassigned before the batch's natural commit point is reached. The new owner starts from the older committed offset and reprocesses records the previous owner already handled successfully — not incorrect under at-least-once semantics and Part 04's idempotency guidance, but unnecessary duplicate work that a revoke-time commit avoids entirely.

on_assign and on_revoke together — the complete rebalance contract

def on_partitions_assigned(consumer, partitions):
    logger.info(f"Assigned: {partitions}")
    # optional: pre-warm any per-partition state here, or explicitly
    # seek to a specific offset if this service needs custom starting logic

def on_partitions_revoked(consumer, partitions):
    logger.info(f"Revoking: {partitions}, committing current progress")
    try:
        consumer.commit(asynchronous=False)
    except Exception as exc:
        logger.error(f"Commit during revoke failed, next owner may reprocess more: {exc}")
    # any per-partition in-memory state (batching buffers, local caches)
    # tied to these specific partitions should be flushed or discarded here

consumer.subscribe(
    ['orders.events'],
    on_assign=on_partitions_assigned,
    on_revoke=on_partitions_revoked,
)

Cooperative rebalancing changes which partitions on_revoke actually sees

Per Module 03's cooperative rebalancing coverage, with the incremental cooperative protocol, on_revoke is called only with the specific partitions actually moving, not the consumer's entire assignment — unaffected partitions are never revoked at all. Code that assumes on_revoke always receives the full current assignment (a holdover assumption from the older eager protocol) can commit or clean up more state than necessary under cooperative rebalancing.

// Part 09 — The production-readiness checklist

Is This Consumer Actually Production-Ready?

Run any new consumer — or a pull request adding one — against this checklist before it ships. Every item traces to a specific failure mode covered earlier in this module.

✓enable.auto.commit=false is set explicitly for any workflow with a real side effect, paired with deliberate manual commits after processing succeeds.
✓The commit granularity (per record vs. per batch) is a conscious choice, with the reprocessing-window tradeoff understood by whoever chose it.
✓Every processing path that writes to a database, calls an external API, or produces to another topic is genuinely idempotent — verified by asking "what happens if this exact record is processed twice."
✓A bounded retry count exists for processing failures, with a dead letter path for records that exhaust it — no unbounded retry loop that can block a partition forever.
✓An on_revoke callback commits current progress before partitions are taken away, not just relying on the normal commit cadence to happen to land in time.
✓Shutdown is wired to real signal handling, calls consumer.wakeup() to interrupt a blocking poll() promptly, and performs a final synchronous commit before consumer.close().
✓max.poll.records and max.poll.interval.ms are sized together against real, measured per-record processing time, not left at defaults and hoped to work under load.
✓isolation.level is set to read_committed whenever upstream producers use Kafka transactions.
✓Consumer lag is monitored per partition, not just as a group-wide total that can hide one badly lagging partition behind many healthy ones.
✓A DLQ record carries enough context (source partition, source offset, error details) to actually diagnose and potentially replay the failure later.

Idempotency is the one item you cannot retrofit casually

Every other item on this checklist can be added to an existing consumer without touching its data model. Idempotent processing often requires a schema change (a unique constraint, an idempotency-key column) or a shift to upsert semantics — design it in from the start rather than treating it as an afterthought once duplicates in production force the issue.

// Part 10 — Testing consumer reliability

Idempotency and Rebalance Safety Are Claims. Test Them.

A consumer's reliability design makes specific, checkable claims: "reprocessing this record twice is harmless," "a rebalance mid-batch never loses committed progress," "a poison message never blocks the partition forever." Each of these should be verified with a targeted test rather than trusted on inspection alone — reliability bugs in consumers are exactly the kind that pass a normal functional test suite and only appear under a crash or a rebalance in production.

Test 1 — processing the same record twice produces the same end state

This is the single highest-value test for any consumer following the idempotent-processing pattern from Part 04: feed the exact same record through the processing function twice and assert the resulting state is identical to processing it once.

an idempotency test that directly exercises the redelivery scenario

def test_processing_same_event_twice_is_idempotent(test_db):
    event = {
        'order_id': 'order-42',
        'event_id': 'evt-9001',
        'status': 'shipped',
        'event_timestamp': '2026-09-01T10:00:00Z',
    }

    consumer = OrderEventConsumer(
        bootstrap_servers='localhost:9092',
        topic='test.orders.events',
        dlq_topic='test.orders.events.dlq',
        group_id='test-group',
    )

    consumer._process_order_event(event)
    state_after_first = test_db.query(
        "SELECT status, updated_at FROM order_status WHERE order_id = %s",
        ('order-42',),
    )

    consumer._process_order_event(event)  # simulate exact redelivery
    state_after_second = test_db.query(
        "SELECT status, updated_at FROM order_status WHERE order_id = %s",
        ('order-42',),
    )

    assert state_after_first == state_after_second
    row_count = test_db.query(
        "SELECT count(*) FROM order_status WHERE order_id = %s", ('order-42',)
    )
    assert row_count[0][0] == 1  # never a duplicate row

Test 2 — an out-of-order redelivery does not regress state

Beyond exact duplication, verify the timestamp-guard pattern from Part 04 actually rejects a stale event replayed after a newer one — a scenario that plain duplicate detection alone would not catch.

verifying the timestamp guard rejects out-of-order redelivery

def test_out_of_order_event_does_not_regress_state(test_db):
    newer_event = {
        'order_id': 'order-42', 'event_id': 'evt-2',
        'status': 'delivered', 'event_timestamp': '2026-09-01T12:00:00Z',
    }
    older_event = {
        'order_id': 'order-42', 'event_id': 'evt-1',
        'status': 'shipped', 'event_timestamp': '2026-09-01T10:00:00Z',
    }

    consumer._process_order_event(newer_event)
    consumer._process_order_event(older_event)  # arrives late, after the newer one

    row = test_db.query(
        "SELECT status FROM order_status WHERE order_id = %s", ('order-42',)
    )
    assert row[0][0] == 'delivered'  # older event must NOT have overwritten this

Test 3 — on_revoke commits before the partition changes hands

Simulate a rebalance mid-batch and confirm the committed offset reflects everything processed before the revoke, not just whatever the normal commit cadence had already landed.

verifying the revoke callback actually commits pending progress

def test_on_revoke_commits_pending_progress(kafka_test_cluster):
    consumer = OrderEventConsumer(
        bootstrap_servers=kafka_test_cluster.bootstrap_servers,
        topic='test.orders.events',
        dlq_topic='test.orders.events.dlq',
        group_id='test-group',
    )

    # process some records without hitting a natural batch-commit boundary
    for msg in kafka_test_cluster.produce_and_fetch(count=10):
        consumer._process_order_event(json.loads(msg.value()))
        # deliberately no commit() call here -- simulating mid-batch state

    # simulate the coordinator revoking this consumer's partitions
    partitions = [TopicPartition('test.orders.events', 0)]
    consumer._on_partitions_revoked(consumer.consumer, partitions)

    committed = consumer.consumer.committed(partitions)
    assert committed[0].offset == 10  # revoke-time commit caught the pending work

Test 4 — a poison message does not block the partition

Inject a record guaranteed to fail processing every time, and confirm the consumer routes it to the DLQ after the configured retry budget and continues processing everything after it, rather than stalling indefinitely.

verifying a poison message is dead-lettered, not blocking

def test_poison_message_routes_to_dlq_and_partition_continues(kafka_test_cluster):
    kafka_test_cluster.produce('test.orders.events', key='bad', value=b'not-valid-json{{{')
    kafka_test_cluster.produce('test.orders.events', key='good', value=json.dumps(
        {'order_id': 'order-99', 'event_id': 'evt-99', 'status': 'created',
         'event_timestamp': '2026-09-01T09:00:00Z'}
    ).encode())

    consumer = OrderEventConsumer(
        bootstrap_servers=kafka_test_cluster.bootstrap_servers,
        topic='test.orders.events',
        dlq_topic='test.orders.events.dlq',
        group_id='test-group-poison',
    )

    # process both records -- the poison one should exhaust retries and
    # route to DLQ, the good one right behind it should still process normally
    processed = consumer.drain_available_records(timeout=10.0)

    dlq_messages = kafka_test_cluster.consume('test.orders.events.dlq', timeout=5.0)
    assert len(dlq_messages) == 1
    assert b'not-valid-json' in dlq_messages[0].value()

    good_state = test_db.query(
        "SELECT status FROM order_status WHERE order_id = %s", ('order-99',)
    )
    assert good_state[0][0] == 'created'  # confirms the partition was not blocked

Run these against a real, local multi-broker cluster, not mocks

Mocking the Kafka client for these specific tests defeats their purpose — the behaviors under test (rebalance timing, actual commit persistence, DLQ delivery) are exactly the parts a mock would fake away. A local Docker-based Kafka cluster in CI, even a single-broker KRaft instance, is worth the setup cost for this category of test.

// Misconceptions

Five Misconceptions About Building Kafka Consumers

✕ ""Manual commit means I need to commit after every single record for safety""

Part 03 shows per-batch commits are the common production pattern — they trade a slightly larger reprocessing window on crash for meaningfully better throughput, and are safe as long as the processing itself is idempotent, per Part 04.

✕ ""If I never auto-commit, I never reprocess a record""

Part 03 and Part 04 are explicit that at-least-once delivery, which manual commit-after-processing still provides, guarantees occasional reprocessing on crash or rebalance regardless of commit strategy. The fix is idempotent processing, not chasing a commit strategy that eliminates redelivery entirely — that would require exactly-once semantics scoped narrowly to Kafka-to-Kafka pipelines.

✕ ""A dead letter queue means that record is fully handled and can be forgotten""

Part 05 frames the DLQ as an escape valve that unblocks the partition, not a resolution. An unmonitored, growing DLQ is silent, permanent data loss for whatever business process depended on that record actually being processed.

✕ ""consumer.close() and process exit are basically the same thing""

Part 07 shows consumer.close() sends an explicit LeaveGroup notice that triggers immediate rebalancing, versus an unclean exit that leaves the group coordinator waiting out the full session.timeout.ms before it notices the consumer is gone — a meaningful difference during a rolling deploy restarting many instances.

✕ ""on_revoke always receives every partition the consumer currently owns""

Part 08 notes that under the cooperative incremental rebalancing protocol, which is now the default in most client versions, on_revoke is called only with the specific partitions actually moving — code written assuming the older eager protocol's full-revocation behavior can commit or clean up more state than the rebalance actually requires.

// What This Looks Like at Work

What This Looks Like on Day One

At Notion: a document-sync consumer occasionally applies the same edit-event twice after a deploy, producing a visibly duplicated change in a document's history. Following Part 04, the on-call engineer finds the processing logic does a plain INSERT keyed on nothing in particular, rather than an upsert or a unique-constraint check on the event's own ID. The fix is exactly the upsert-with-timestamp-guard pattern from Part 04 — no change to the deployment or commit strategy required, because the redelivery itself was never the actual bug.

At Instacart: a shopper-assignment consumer group experiences a rebalance during every single rolling deploy, and each rebalance takes the full session.timeout.ms to resolve, adding visible delay to an otherwise routine deploy. Per Part 07, the team discovers the service was never calling consumer.close() on shutdown — the container's SIGTERM handler stopped the poll loop but exited before the consumer sent an explicit departure notice to the group coordinator. Wiring consumer.wakeup() and a proper close()call into the shutdown sequence turns each deploy's rebalance from a multi-second stall into a near-instant handoff.

At Brex: a transaction-categorization consumer keeps a single partition's DLQ quietly filling up for two weeks before anyone notices, because there was no alert wired to DLQ message volume — just a dashboard nobody was actively watching. Per Part 05 and Part 09's checklist, the fix is not just draining the backlog; it's adding an alert on DLQ depth so this specific gap — "the escape valve is not the same as a resolution" — cannot silently recur.

// Interview Prep

5 Interview Questions — With Complete Answers

Q1. Explain why "commit after processing" is not, by itself, a complete reliability strategy for a Kafka consumer.

Committing after processing, rather than before or on an independent timer, is the right foundation — it avoids the specific auto-commit failure mode from Module 03 where a crash between an early commit and actual work completion silently skips records. But per Part 03 and Part 04, it only prevents skipped work; it does not prevent duplicate work. If a consumer processes a record successfully, has a real side effect like a database write or an external API call, and then crashes before the commit itself lands — or a rebalance takes the partition away at exactly that moment — the same record gets redelivered to whoever processes that partition next. Commit-after-processing guarantees at-least-once delivery, not exactly-once. The complete strategy is commit-after-processing plus idempotent processing logic, so that the redelivery Part 03's pattern still allows is harmless when it happens. I'd point to the upsert-with-timestamp-guard pattern from Part 04 as the concrete mechanism — it makes reprocessing the same event, or even an out-of-order one, a no-op rather than a source of corrupted state.

Q2. How do you handle a message that fails processing every single time you attempt it, without stopping the rest of the partition?

This is the poison message problem from Part 05. Because a consumer following the commit-after-processing pattern only advances its offset after success, a message that always fails would block every record behind it indefinitely if retried without bound. The fix is a bounded retry count — enough attempts to absorb a genuinely transient failure, typically with a short backoff between attempts — followed by routing the message to a dead letter topic once that budget is exhausted, and then committing past it anyway. The DLQ record should carry the original payload, the error details, and the source partition and offset, so an engineer can later diagnose the root cause and, once it's fixed, replay the message back through the original pipeline. The subtlety I'd flag: the function handling this should return success to the caller in both the "processed cleanly" case and the "exhausted retries, routed to DLQ" case — both are handled from the offset-commit perspective. The distinction between them lives in monitoring and in the DLQ itself, not in whether the partition keeps moving.

Q3. What is the purpose of an on_revoke / onPartitionsRevoked callback, and what happens if a consumer doesn't implement one?

Per Part 08, on_revoke fires immediately before the group coordinator reassigns a consumer's partitions to someone else during a rebalance. It is the last opportunity for the outgoing consumer to commit any offsets for work it has already processed but not yet committed through its normal commit cadence. Without it, a consumer relying solely on periodic or per-batch commits can lose that opportunity at exactly the worst moment — a rebalance triggered mid-batch means the current batch's successfully processed records never get committed, because the partition changes hands before the batch's natural commit point arrives. The new owner then starts from an older committed offset and reprocesses records the previous owner already handled. This isn't a correctness bug if the processing is idempotent, per Part 04 — but it's unnecessary duplicate work, and at scale, across every rebalance a busy consumer group experiences, that adds up. Implementing on_revoke to do one final synchronous commit closes that gap directly.

Q4. Why does a consumer need consumer.wakeup() during shutdown instead of just checking a flag at the top of the loop?

Because poll() itself blocks for up to its configured timeout waiting for records, per Part 07. If a shutdown signal arrives while the consumer is inside a blocking poll() call, simply setting a flag doesn't help until that call returns on its own — which could be up to the full timeout later, and that delay compounds with whatever other shutdown work needs to happen within the orchestrator's grace period. consumer.wakeup(), called from the signal handler itself, interrupts a blocking poll() call immediately by raising inside it, so the main loop's exit condition gets checked right away rather than waiting out the timeout. It's a small detail, but it's the difference between a shutdown that reliably completes within a tight grace period and one that occasionally races a SIGKILL, especially under load where the loop might be doing other work between poll() calls too. I'd also mention that the full shutdown sequence matters: after the loop exits, a final synchronous commit and an explicit consumer.close() call are what let the group coordinator rebalance immediately, rather than waiting out session.timeout.ms for a departure it doesn't yet know happened.

Q5. Walk me through how you would design idempotent processing for a consumer that updates an order's current status in a database.

Since the event represents a state transition rather than an append-only fact, per Part 04 I'd reach for an upsert keyed on the order's ID rather than trying to detect and skip duplicates explicitly. An INSERT ... ON CONFLICT DO UPDATE keyed on order_id means processing the exact same status update twice leaves the row in the same final state both times — idempotent by construction, with no separate duplicate-tracking table or check required. I'd go one step further and add a guard clause inside the update — only apply the new status if the incoming event's timestamp is newer than what's already stored. That protects against a second failure mode beyond plain duplication: out-of-order redelivery, where an older event gets reprocessed after a newer one has already landed, which a naive upsert without the timestamp guard would incorrectly let overwrite the more current state. For the parts of the pipeline where the side effect is external rather than a database row — say, this event also needs to notify a payment provider — I'd use a stable, event-assigned idempotency key on that external call as well, since a database-level upsert doesn't protect an API call made outside that transaction boundary.

// Common Mistakes

The Mistakes That Make Kafka Consumers Unreliable in Production

Leaving enable.auto.commit=true on a consumer with real side effects

Part 02 and Part 03 both point to this as the single most common source of silent skipped work — the commit timer runs independent of whether your application has actually finished processing.

Writing processing logic that is not idempotent, then treating every duplicate as a surprise

At-least-once delivery guarantees occasional reprocessing regardless of commit strategy, per Part 04. Duplicates are not a bug in Kafka to chase — they are an expected condition the processing logic must be designed to absorb.

Retrying a failing record indefinitely instead of routing to a DLQ after a bounded number of attempts

Part 05 shows this blocks every record behind the poison message on that partition, potentially forever, since the consumer never advances its committed offset past it.

Skipping the on_revoke callback and relying only on normal commit cadence

Part 08 shows this can lose the chance to commit already-processed work right before a rebalance takes the partition away, causing unnecessary reprocessing by the next owner even when the processing itself was idempotent.

Killing the consumer process without calling consumer.close()

Part 07 is explicit that close() sends an explicit departure notice that lets the group rebalance immediately; without it, the coordinator waits out the full session.timeout.ms before reassigning the abandoned partitions.

// Error Library

Errors You Will Hit — And Exactly Why They Happen

Consumer group rebalances repeatedly with logs showing CommitFailedException or a similar "commit cannot be completed" error

Cause: Per Part 02, this almost always means max.poll.interval.ms was exceeded before the commit call was reached — the consumer was removed from the group mid-processing, and the offset it is trying to commit is for a partition it no longer owns.

Fix: Compare real per-record processing time times max.poll.records against max.poll.interval.ms, exactly as covered for producers' consumer counterpart in Module 03. Lower max.poll.records, raise max.poll.interval.ms to match realistic processing time, or move slow work off the poll thread.

The same order status update appears to apply twice in application logs, but the database ends up in the correct final state either way

Cause: This is at-least-once delivery working as expected, per Part 04 — a crash or rebalance caused reprocessing of an already-handled record. The database ending up correct means the idempotent upsert pattern is doing its job.

Fix: This is not a bug to fix; it is confirmation the idempotency pattern is correct. If it were instead producing an incorrect final state, that would indicate a missing timestamp guard or unique constraint, not a delivery-layer problem.

DLQ topic volume grows steadily over days with no corresponding alert firing

Cause: Per Part 05 and Part 09, a DLQ with no monitoring wired to its message rate or depth is invisible by design — the whole point of the pattern is to unblock the partition, which it does silently whether or not anyone is watching.

Fix: Add an alert on DLQ message rate or depth as a first-class part of shipping any consumer that uses this pattern, not as a follow-up task — Part 09's checklist treats this as a hard requirement, not a nice-to-have.

A newly deployed consumer instance in a rolling deploy causes a multi-second processing pause across the whole group

Cause: Per Part 07, the previous instance of this consumer likely did not call consumer.close() cleanly on shutdown, leaving the group coordinator to wait out the full session.timeout.ms before recognizing the departure and reassigning partitions.

Fix: Wire consumer.wakeup() into signal handling and ensure consumer.close() is reliably called in a finally block, sending an explicit LeaveGroup notice that lets the group rebalance immediately instead of waiting out the timeout.

A consumer reading from a topic written by a transactional producer occasionally processes records that later disappear from downstream reporting

Cause: Per Part 02, isolation.level was left at its default of read_uncommitted, which exposes records from transactions that were later aborted — the consumer processed a write that was never actually meant to be visible.

Fix: Set isolation.level=read_committed on any consumer reading from a topic written by a transactional producer, so only committed transaction results are ever visible to this consumer.

🎯 Key Takeaways

✓group.id, auto.offset.reset, and enable.auto.commit are not boilerplate — each is a deliberate decision about fan-out, replay behavior on first startup, and when work actually counts as done.
✓Commit-after-processing (per record or per batch) prevents silently skipped work, but it guarantees at-least-once delivery, not exactly-once — idempotent processing is what makes the resulting redelivery harmless.
✓A bounded retry count followed by a dead letter queue is what stops one poison message from permanently blocking every record behind it on the same partition — and DLQ volume needs its own monitoring, not just its existence.
✓on_revoke / onPartitionsRevoked is the last chance to commit already-processed work before a rebalance hands a partition to another consumer; skipping it causes unnecessary (though usually harmless, if idempotent) reprocessing.
✓consumer.wakeup() plus a proper consumer.close() in shutdown lets a consumer interrupt a blocking poll() promptly and notify the group coordinator explicitly, turning a rebalance from a session.timeout.ms-long stall into a near-instant handoff.
✓max.poll.records and max.poll.interval.ms must be sized together against real, measured per-record processing time — including any retry backoff time — not left at defaults and hoped to work under production load.

What to learn next

Kafka Connect

Apache Kafka · 23 min · +150 XP

Stream Processing and Kafka Streams

Apache Kafka · 26 min · +150 XP

Producer Design

Kafka Connect

Discussion

Have a better approach? Found something outdated? Share it — your knowledge helps everyone learning here.

Continue with GitHub