Implement an idempotent event consumer

Expected question

"Implement an idempotent event consumer: given at-least-once delivery, process each event_id at most once for side effects, and advance the offset only after successful handling."

Variant forms

• "Consumers see duplicates — how do you make handlers safe?"
• "Implement process(event) with a processed-ids store."
• "Commit offsets before or after side effects — which is correct for at-least-once?"
• "What happens if we crash after the side effect but before offset commit?"
• "Add a poison-message path that skips and dead-letters."
• "Make the dedupe store safe under concurrent partition handlers."
• "How is this different from Kafka exactly-once transactions?"
• "Implement a tiny in-memory broker + consumer group to demo the contract."

The question, as it might actually be asked

You receive events with unique event_ids from an at-least-once log (Kafka-style). Implement handle(event) -> HandleResult that applies a side effect at most once per event_id, records progress, and distinguishes success, duplicate, and poison failures. Define when the caller may advance the offset.

How you'd talk while coding

I'd start with the delivery contract: at-least-once is the default, so duplicates will happen. Clarify whether event_id is globally unique, in-memory vs DB dedupe for the interview, retryable vs poison, and concurrency. Brute always calls apply — double charges. Correct: check processed set, apply, then record success; only then allow offset advance. Retryable errors leave advance_offset=False. Poison → DLT + mark processed + advance (skip). I'd name the crash window out loud: die after side effect before recording and at-least-once redelivers — production pairs side effect + inbox in one transaction. Staff+ stop line: "Kafka EOS helps consume→produce atomicity; it does not make SMTP or card charges exactly-once. Idempotency still required. I won't design the cluster."

Where this actually gets asked

Coding / LLD bridge for event-driven systems. Interviewers want at-least-once + idempotency (the production default), not hand-waving "exactly-once." Pairs with general-system-design/18 and 09.

Problem

Implement an idempotent consumer helper:

• handle(event: Event) -> HandleResult where Event has event_id and payload
• Side effect is provided as apply(payload) -> None (may raise)
• Duplicates must not re-apply side effects
• Poison / non-retryable errors can be marked skipped for offset advance + dead-letter hook
• Retryable errors must not advance progress

Clarifying questions you should ask first

1. At-least-once, at-most-once, or transactional EOS assumed?
2. Is event_id globally unique, or unique per aggregate/stream?
3. Is the dedupe store in-memory for the interview, or should we narrate a DB unique constraint?
4. What exceptions are retryable vs poison?
5. Single-threaded handler or concurrent handlers on different partitions?

Approach ladder

Reference solution (Python)

from __future__ import annotations

from dataclasses import dataclass
from enum import Enum
from threading import Lock
from typing import Callable, Protocol, Set


class HandleStatus(str, Enum):
    APPLIED = "applied"
    DUPLICATE = "duplicate"
    RETRY = "retry"
    POISON_SKIPPED = "poison_skipped"


@dataclass(frozen=True)
class Event:
    event_id: str
    payload: str


@dataclass(frozen=True)
class HandleResult:
    status: HandleStatus
    advance_offset: bool


class PoisonMessageError(Exception):
    """Non-retryable handler failure; safe to skip after dead-lettering."""


class RetryableError(Exception):
    """Transient failure; do not advance offset."""


class DeadLetter(Protocol):
    def send(self, event: Event, reason: str) -> None: ...


class IdempotentConsumer:
    """At-least-once safe helper: side effects once per event_id."""

    def __init__(
        self,
        apply: Callable[[str], None],
        dead_letter: DeadLetter | None = None,
        processed: Set[str] | None = None,
    ) -> None:
        self._apply = apply
        self._dead_letter = dead_letter
        self._processed: Set[str] = processed if processed is not None else set()
        self._in_flight: Set[str] = set()
        self._lock = Lock()

    def handle(self, event: Event) -> HandleResult:
        with self._lock:
            if event.event_id in self._processed:
                return HandleResult(HandleStatus.DUPLICATE, advance_offset=True)
            if event.event_id in self._in_flight:
                # Another worker is applying; caller should not advance (wait / retry delivery).
                return HandleResult(HandleStatus.RETRY, advance_offset=False)
            self._in_flight.add(event.event_id)

        try:
            self._apply(event.payload)
        except PoisonMessageError as exc:
            if self._dead_letter is not None:
                self._dead_letter.send(event, str(exc))
            with self._lock:
                self._in_flight.discard(event.event_id)
                self._processed.add(event.event_id)
            return HandleResult(HandleStatus.POISON_SKIPPED, advance_offset=True)
        except RetryableError:
            with self._lock:
                self._in_flight.discard(event.event_id)
            return HandleResult(HandleStatus.RETRY, advance_offset=False)
        except Exception:
            with self._lock:
                self._in_flight.discard(event.event_id)
            # Unknown errors: fail closed — do not commit progress.
            return HandleResult(HandleStatus.RETRY, advance_offset=False)

        with self._lock:
            self._in_flight.discard(event.event_id)
            # Crash window: if we die here, at-least-once redelivers; durable inbox/unique
            # constraint with the side effect closes this in production (Staff+ narrative).
            self._processed.add(event.event_id)
        return HandleResult(HandleStatus.APPLIED, advance_offset=True)

Complexity: O(1) average per event with a hash set; memory grows with distinct event_ids unless bounded or externalized.

Verbal tests to narrate

1. First delivery applies side effect and returns advance_offset=True.
2. Redelivery of same event_id returns DUPLICATE, does not call apply again.
3. RetryableError → advance_offset=False so the log redelivers.
4. PoisonMessageError → dead-letter once, mark processed, advance offset (skip).
5. Concurrent duplicate deliveries: only one apply wins (lock / unique constraint).

Staff+ deep dive

What's expected at each level

• Mid-level: Checks a set before applying; explains duplicates from retries.
• Senior: Ties offset commit to success; separates retryable vs poison; writes clear tests.
• Staff+: Names the crash window, transactional inbox/outbox pairing, and why EOS ≠ no idempotency.
• Principal: Sets org standards for event_id, DLT ownership, and consumer lag / poison SLOs.

Follow-up questions to expect

• "Commit sync vs async?" — Async risks more redelivery; sync after batch cuts duplicates at latency cost. I'd pick based on duplicate cost.
• "Dedupe without storing every id forever?" — Upsert natural business keys; compacted inbox by aggregate version; TTL only when re-apply is safe.
• "Show the outbox on the producer?" — Same interview often pivots to dual-write; see design entry 18.

Related

• 09-design-inmemory-pubsub.md
• 04-concurrent-bounded-queue.md
• 18-circuit-breaker.md
• ../general-system-design/18-event-driven-architecture-with-kafka.md
• 00-staff-plus-coding-bar.md