Building an Idempotent Consumer with a Processed-Events Table

The processed-events table is the most portable deduplication pattern in event processing: it works with any broker, needs no broker-specific transaction support, and puts the guarantee in the same database as the data it protects. This runbook implements it end to end and sits under idempotent consumer patterns for event streams.

Prerequisites. You need a relational store that holds the business data — the whole point is co-locating the marker with the effect — and an understanding of why at-least-once delivery makes duplicates inevitable. Familiarity with database unique constraints helps, since the whole mechanism is one.

Step 1 — Create the table

CREATE TABLE processed_events (
    consumer_group text        NOT NULL,
    event_id       text        NOT NULL,
    topic          text        NOT NULL,
    partition      int         NOT NULL,
    "offset"       bigint      NOT NULL,
    processed_at   timestamptz NOT NULL DEFAULT now(),
    PRIMARY KEY (consumer_group, event_id)
);

-- Pruning scans by time; a dedicated index keeps the reaper off the primary key.
CREATE INDEX processed_events_processed_at ON processed_events (processed_at);

The primary key is (consumer_group, event_id), and both halves earn their place. event_id is producer-assigned and stable across redelivery, which the broker offset is not — offsets restart at zero when a topic is recreated and mean nothing after a partition reassignment. consumer_group allows two independent consumers of the same topic to write markers into one database without shadowing each other.

topic, partition and offset are stored but not part of the key. They exist for debugging: when a duplicate shows up, the first question is always “which partition did the redelivery come from”, and reconstructing that from broker logs afterwards is painful.

Processed-events table absorbing a redelivery A broker partition delivers event evt-42 twice. The first delivery opens a transaction that inserts a marker row into processed_events and updates the ledger, then commits. The second delivery opens a transaction whose marker insert hits the unique constraint, so the whole transaction rolls back and the ledger is untouched. The consumer acknowledges both deliveries. partition 3 evt-42 delivery 1 delivery 2 BEGIN INSERT processed_events UPDATE ledger COMMIT ✓ BEGIN INSERT → unique violation ledger never touched ROLLBACK ledger one entry for evt-42

Step 2 — Insert the marker and apply the effect in one transaction

This is the entire pattern. Everything else is operational detail.

from psycopg.errors import UniqueViolation

def handle(event, conn) -> str:
    try:
        with conn.transaction():
            conn.execute(
                """INSERT INTO processed_events
                       (consumer_group, event_id, topic, partition, "offset")
                   VALUES (%s, %s, %s, %s, %s)""",
                (GROUP, event.id, event.topic, event.partition, event.offset),
            )
            apply_effect(event, conn)     # every write here shares the transaction
        return "processed"
    except UniqueViolation:
        return "duplicate"                # already applied — safe to acknowledge
func Handle(ctx context.Context, db *sql.DB, ev Event) (string, error) {
    tx, err := db.BeginTx(ctx, nil)
    if err != nil { return "", err }
    defer tx.Rollback()

    _, err = tx.ExecContext(ctx,
        `INSERT INTO processed_events (consumer_group, event_id, topic, partition, "offset")
         VALUES ($1,$2,$3,$4,$5)`, group, ev.ID, ev.Topic, ev.Partition, ev.Offset)
    if isUniqueViolation(err) { return "duplicate", nil }   // rollback via defer
    if err != nil { return "", err }

    if err := applyEffect(ctx, tx, ev); err != nil { return "", err }
    return "processed", tx.Commit()
}

Two rules make this correct rather than merely plausible. The insert comes first, so a duplicate aborts before any work is done — putting it last means the effect runs and is then rolled back, which is wasteful and, if the effect touches anything non-transactional, wrong. And apply_effect must use the same connection and transaction; a handler that grabs a fresh connection from a pool has silently opted out of the guarantee.

Step 3 — Commit the offset after the transaction

The order is: database transaction commits, then broker offset commits. Reversing it converts a crash into data loss instead of a harmless duplicate.

for msg in consumer:                      # enable_auto_commit=False
    event = decode(msg)
    outcome = handle(event, conn)
    consumer.commit()                     # only after handle() returned
    metrics.inc("events_handled_total", outcome=outcome)

If the process dies between the two commits, the event is redelivered, the marker insert hits the unique violation, and the second attempt is a no-op. That is the whole point: the pattern makes an offset-commit crash boring.

Step 4 — Prune without blocking

Markers accumulate at the event rate and must be pruned, but a naive DELETE ... WHERE processed_at < now() - interval '8 days' on a 400-million-row table takes a lock long enough to stall the consumer.

-- Batched delete: bounded work per statement, no long-held locks.
DELETE FROM processed_events
 WHERE ctid IN (
     SELECT ctid FROM processed_events
      WHERE processed_at < now() - interval '8 days'
      LIMIT 10000
 );
-- Run in a loop until zero rows are affected, sleeping 200 ms between batches.

Retention must exceed the broker’s. If Kafka keeps 7 days and markers are pruned at 3, a consumer reset to the earliest offset replays four days of events whose markers are gone, and every one of them is applied a second time. Eight days against a seven-day topic is a reasonable margin.

For very high event rates, partition the table by day and drop whole partitions instead:

CREATE TABLE processed_events (
    consumer_group text NOT NULL,
    event_id       text NOT NULL,
    processed_at   timestamptz NOT NULL DEFAULT now(),
    PRIMARY KEY (consumer_group, event_id, processed_at)
) PARTITION BY RANGE (processed_at);

CREATE TABLE processed_events_2026_08_01 PARTITION OF processed_events
    FOR VALUES FROM ('2026-08-01') TO ('2026-08-02');

-- Pruning becomes O(1): DROP TABLE processed_events_2026_07_24;

The cost is that processed_at must join the primary key, which weakens the uniqueness guarantee across partition boundaries — an event redelivered a day later would not collide. Accept it only when the redelivery window is provably shorter than one partition.

Step 5 — Verify across a rebalance

Rebalance behaviour with a processed-events table A timeline in three phases. In phase one consumer A owns partition 3, commits its database transaction for event forty-two, and dies before committing the broker offset. In phase two the group rebalances and partition 3 is reassigned to consumer B. In phase three consumer B receives event forty-two again, its marker insert hits the unique constraint, and it acknowledges without reapplying the effect. consumer A owns p3 DB commit ✓ evt-42 dies before offset commit rebalance p3 → consumer B consumer B replays evt-42 unique violation → rollback ledger unchanged effect applied once offset rewinds duplicate absorbed Net result: one ledger entry, two deliveries, zero operator involvement. 
# Force the scenario in a test environment: kill the consumer between commits.
kafka-console-producer --topic payments --property parse.key=true \
  <<< 'evt-42:{"amount":4500,"account":"acct_7"}'

# Kill consumer A mid-flight, then let the group rebalance.
kill -9 "$CONSUMER_A_PID"
kafka-consumer-groups --describe --group ledger-writer   # watch p3 reassign

psql -c "SELECT count(*) FROM ledger_entries WHERE event_id='evt-42'"   # expect 1
psql -c "SELECT count(*) FROM processed_events WHERE event_id='evt-42'" # expect 1

Failure scenarios and debugging

Failure Scenario Remediation Steps Observability Hooks
Handler opens its own database connection instead of using the transaction’s, so the marker and the effect commit independently Pass the transaction handle into apply_effect and forbid connection acquisition below the handler boundary — a lint rule or a runtime assertion on connection identity catches it. processed_events count versus effect-row count per day; any divergence proves the writes are not sharing a transaction
Broker offset committed before the database transaction, so a crash silently drops the event Reorder to database-then-offset and disable auto-commit explicitly. Assert enable.auto.commit=false at startup and refuse to start otherwise. events_lost_total from a producer-side sequence-gap detector; consumer lag that never recovers after a restart
Markers pruned at 3 days against a 7-day topic; a consumer reset replays 4 days of events and reapplies every one Set marker retention above broker retention and assert the relationship at startup by reading the topic config. processed_events_oldest_age_seconds gauge compared against kafka_topic_retention_seconds; alert if the ratio drops below 1.1
Table grows to 400 million rows and the unaudited DELETE locks the table for minutes, stalling the consumer Switch to the batched delete above, or partition by day and drop partitions. Never run an unbounded DELETE on the marker table. processed_events_row_count gauge; consumer_lag spikes correlated with the pruning job’s schedule
Producer reuses an event id across two semantically different events, so the second is silently dropped as a duplicate Treat event ids as immutable and unique per event at the producer, generated with a ULID or UUIDv7. Add a producer-side assertion that an id is never reused within a session. events_handled_total{outcome="duplicate"} — a rate above the expected redelivery baseline points at producer id reuse, not broker redelivery
Marker retention must outlast broker retention Two bars over the same time axis. Broker retention runs to seven days. Marker retention set to three days ends early, leaving a four-day region in which a consumer reset to the earliest offset replays events whose markers have been deleted. An eight-day marker bar covers the broker fully with a one-day margin. broker 7-day topic retention markers (3 d) replays here reapply every effect markers (8 d) covers the broker with a day to spare

SRE / observability checklist

  1. events_handled_total{outcome} — Counter labelled processed or duplicate. The duplicate ratio is the health signal: a steady 0.01%–0.5% is normal at-least-once behaviour, and a jump means a rebalance loop or a producer bug.
  2. processed_events_row_count — Gauge from a periodic count. Alert on growth that outpaces the pruning rate, which predicts the lock incident before it happens.
  3. processed_events_oldest_age_seconds — Gauge of the oldest surviving marker. Must stay above the broker’s retention; alert if the ratio falls below 1.1.
  4. event_handle_duration_seconds — Histogram of the whole transaction. A p99 above 200 ms usually means the marker index has bloated and needs a REINDEX CONCURRENTLY.
  5. consumer_offset_commit_lag_seconds — Gauge of the gap between database commit and offset commit. A widening gap means crashes in that window are becoming more likely, not less.
  6. event_id and consumer_group log fields — On every handled event, so a duplicate can be traced back to the partition and offset that redelivered it.