Designing Idempotent Kafka Consumers with Per-Message TTL Dedup Keys

I ran into a weird failure mode in a system that used Kafka for events and a relational database for persistence: duplicate orders sometimes got processed again hours after the original event, even though the consumer was “idempotent.” The root cause wasn’t idempotency itself—it was how I derived the deduplication key and when I expired it.

This post is about a very specific architecture trade-off I ended up nailing down: per-message TTL dedup keys for Kafka consumers, so “exactly-once-ish” behavior survives restarts and doesn’t keep dedup records forever.

---

The failure mode I saw

My consumer did this (simplified):

1. Read a message from Kafka.
2. Write to a database table.
3. Mark that message as “processed” in a dedup table.
4. Commit the Kafka offset.

On paper that’s idempotent: if the same message arrives again, the consumer checks dedup and skips the write.

But in practice:

• The consumer restarted a few hours later.
• Kafka redelivered some older messages (not uncommon during rebalances or offset mismanagement).
• My dedup keys had expired too early (or didn’t exist after pruning).
• So those messages were treated as “new” and processed again—creating duplicate rows and inconsistent downstream state.

The trade-off became: How long should I keep dedup entries, and how should I compute keys?

---

The niche trade-off: TTL dedup keys derived from message identity

I eventually settled on this pattern:

• Dedup is keyed by a stable “message identity” (usually a event_id).
• Each dedup record gets a processed_until timestamp (a TTL).
• The consumer treats a message as already processed only if the dedup record exists and hasn’t expired.

This makes behavior predictable across restarts:

• If the message is redelivered within TTL: skip safely.
• If it’s redelivered after TTL: treat as new (because the system intentionally forgot).

Why this is an architecture trade-off

• Long TTL reduces duplicates but causes unbounded growth in the dedup table.
• Short TTL bounds storage but increases the chance that late redeliveries will duplicate work.
• Key correctness matters: if the dedup key changes across retries (or across producers), idempotency silently fails.

---

A concrete implementation (Kafka consumer + Postgres dedup with TTL)

Below is a working example using:

• Python with kafka-python (for Kafka consumption)
• PostgreSQL with psycopg (for database operations)

Data model

I keep a processed_dedup table with:

• event_id: stable identity coming from the producer
• processed_until: expiration time for the dedup record

-- schema.sql
create table if not exists processed_dedup (
  event_id text primary key,
  processed_until timestamptz not null
);

create table if not exists orders (
  order_id text primary key,
  status text not null,
  last_updated_at timestamptz not null default now()
);

The key idea in code

I use an atomic insert to claim the message identity:

• If insert succeeds: this is the first processing attempt within TTL window.
• If it fails due to a duplicate key: then I check expiry.
  • If expired: I delete the old record and insert again.
  • If not expired: skip the message.

This avoids race conditions where two consumer instances might process the same event around the same time.

---

End-to-end consumer code

# consumer.py
import json
from datetime import datetime, timedelta, timezone

from kafka import KafkaConsumer
import psycopg

DB_DSN = "postgresql://postgres:postgres@localhost:5432/app"
TOPIC = "orders"
GROUP_ID = "orders-consumer-v1"

# Keep dedup records for this long.
# Architecture trade-off: longer reduces duplicates, shorter saves storage.
DEDUP_TTL_SECONDS = 4 * 60 * 60  # 4 hours


def now_utc():
    return datetime.now(timezone.utc)


def parse_event(msg_value: bytes) -> dict:
    payload = json.loads(msg_value.decode("utf-8"))
    # Expected shape from producer:
    # {
    #   "event_id": "uuid-or-stable-string",
    #   "order_id": "order-123",
    #   "status": "PAID"
    # }
    return payload


def should_skip_or_claim(cur, event_id: str):
    """
    Try to atomically create a dedup entry.
    If it already exists and isn't expired, we skip.
    If it exists but is expired, we allow re-processing (claim).
    """
    processed_until = now_utc() + timedelta(seconds=DEDUP_TTL_SECONDS)

    # Attempt insert. If event_id exists, we handle it in the exception path.
    try:
        cur.execute(
            """
            insert into processed_dedup (event_id, processed_until)
            values (%s, %s)
            """,
            (event_id, processed_until),
        )
        return "claimed"
    except psycopg.errors.UniqueViolation:
        # event_id already exists; check expiry
        cur.execute(
            """
            select processed_until
            from processed_dedup
            where event_id = %s
            """,
            (event_id,),
        )
        row = cur.fetchone()
        if not row:
            # Extremely rare race; treat as claim and retry
            cur.execute(
                """
                insert into processed_dedup (event_id, processed_until)
                values (%s, %s)
                """,
                (event_id, processed_until),
            )
            return "claimed"

        (existing_processed_until,) = row
        if existing_processed_until > now_utc():
            return "skip"
        else:
            # Expired: remove and re-insert (claim again)
            cur.execute(
                "delete from processed_dedup where event_id = %s",
                (event_id,),
            )
            cur.execute(
                """
                insert into processed_dedup (event_id, processed_until)
                values (%s, %s)
                """,
                (event_id, processed_until),
            )
            return "claimed"


def upsert_order(cur, order_id: str, status: str):
    # This is a separate idempotency layer at the domain level.
    # order_id primary key ensures duplicates can't create multiple rows.
    cur.execute(
        """
        insert into orders (order_id, status, last_updated_at)
        values (%s, %s, now())
        on conflict (order_id)
        do update set
          status = excluded.status,
          last_updated_at = now()
        """,
        (order_id, status),
    )


def main():
    consumer = KafkaConsumer(
        TOPIC,
        bootstrap_servers=["localhost:9092"],
        group_id=GROUP_ID,
        enable_auto_commit=False,  # commit manually after successful DB work
        auto_offset_reset="earliest",
        value_deserializer=lambda v: v,
    )

    with psycopg.connect(DB_DSN) as conn:
        conn.autocommit = False  # explicit transactions
        cur = conn.cursor()

        for msg in consumer:
            event = parse_event(msg.value)
            event_id = event["event_id"]
            order_id = event["order_id"]
            status = event["status"]

            # One transaction covers: dedup decision + domain write + commit offset.
            try:
                action = should_skip_or_claim(cur, event_id)
                if action == "skip":
                    conn.commit()
                    # Commit Kafka offset after transaction
                    consumer.commit()
                    continue

                # If claimed, perform the domain write
                upsert_order(cur, order_id, status)

                conn.commit()
                consumer.commit()

            except Exception:
                conn.rollback()
                # Do not commit offset on failure; message will be retried.
                raise


if ____ == "__main__":
    main()

---

Step-by-step: what happens when a message arrives

Case A: first time within TTL

1. Consumer receives message with event_id = A.
2. should_skip_or_claim() tries to insert into processed_dedup.
3. Insert succeeds → action is "claimed".
4. It upserts into orders.
5. Transaction commits and Kafka offset commits.

Case B: duplicate redelivery within TTL

1. Consumer receives same event_id = A again.
2. Insert fails with UniqueViolation (primary key conflict).
3. It reads processed_until.
4. Since processed_until > now(), action is "skip".
5. No domain write happens.
6. Offset commits (because the message is effectively already handled).

Case C: redelivery after TTL

1. Consumer receives same event_id = A after hours.
2. Insert fails with UniqueViolation.
3. It reads processed_until.
4. If processed_until <= now(), the dedup record is expired.
5. It deletes and re-inserts the dedup record to “claim” the message.
6. It re-upserts the domain record.

In many systems, that last step is acceptable: after TTL, the system intentionally forgets that it processed the message.

---

Choosing a TTL: the practical way I picked 4 hours

I didn’t choose TTL based on theory—I picked it based on observed redelivery windows in my environment:

• Kafka rebalances and consumer restarts produced occasional duplicates.
• The worst-case “late delivery” I observed during testing and incident retrospectives clustered around a couple of hours.
• I added slack for messy operational reality (deploys, flaky networks, backlogs).

The real architecture takeaway: TTL should reflect the time horizon where “duplicate processing is still worse than storage growth.”

---

Key derivation: the hidden foot-gun

One more trade-off bit me hard: the producer must provide a stable event_id.

If I used something like:

• Kafka offset,
• database generated id,
• or a timestamp rounded to seconds,

then retries could create a different event_id, breaking dedup entirely. The dedup table would grow forever and duplicates would slip through.

So my rule became:

• event_id is generated once at the source of truth (the producer), and it must remain stable for the life of that logical event.

---

Conclusion

I learned that “idempotent consumers” aren’t just a boolean property—you end up making explicit architecture trade-offs about dedup key stability and dedup retention time. Using per-message TTL dedup keys with an atomic claim in the database gave me predictable behavior across restarts: duplicates were skipped safely within the TTL window, and after TTL the system intentionally allowed reprocessing without letting dedup storage grow without bounds.