Backfilling a Kafka topic without double-counting

Last month I had to re-process about six days of an events topic. A consumer had been silently dropping events with a negative amount — refunds — because of a filter that was meant to log a warning and instead continued. Those events feed a table of daily counts that a finance team looks at every Monday morning. This is the note I wish I'd had on that Monday.

The trap

The consumer's write looked like this:

INSERT INTO daily_counts (day, customer_id, n)
VALUES ($1, $2, 1)
ON CONFLICT (day, customer_id)
DO UPDATE SET n = daily_counts.n + 1;

That's an increment, and increments aren't idempotent: put the same event through twice and you get two. Kafka gives you at-least-once delivery by default, and you will see an event twice eventually — on a rebalance, on a crash between the database commit and the offset commit, and on any deliberate replay like the one I needed. So the consumer was already subtly wrong before the backfill; the backfill would just make it wrong faster.

What doesn't work

• Resetting the group's offsets to six days ago with kafka-consumer-groups --reset-offsets --to-datetime. This replays everything, including the events that were counted. With increments you now double-count those.
• Deleting the six days of counts and replaying. Tempting, and it would have worked if this topic were the only thing writing to daily_counts. It wasn't; a second consumer on a different topic increments the same rows. (Two writers to one aggregate is its own mistake. Separate note.)
• Consuming into a scratch table and adding the deltas. You still have to know which events were already counted, which is the original problem wearing a hat.

What did work: make the write idempotent, then replay

Step 1: land every event under a unique key. Every event already carried an event_id (a UUID from the producer). Record it:

CREATE TABLE events_seen (
  event_id    uuid        NOT NULL,
  day         date        NOT NULL,
  customer_id bigint      NOT NULL,
  seen_at     timestamptz NOT NULL DEFAULT now(),
  PRIMARY KEY (event_id, day)
) PARTITION BY RANGE (day);

Step 2: only count what you actually inserted.

WITH ins AS (
  INSERT INTO events_seen (event_id, day, customer_id)
  VALUES ($1, $2, $3)
  ON CONFLICT (event_id, day) DO NOTHING
  RETURNING day, customer_id
)
INSERT INTO daily_counts (day, customer_id, n)
SELECT day, customer_id, 1 FROM ins
ON CONFLICT (day, customer_id)
DO UPDATE SET n = daily_counts.n + 1;

If the event has been seen before, ins returns no rows and the second insert does nothing. It's one statement, so one transaction: a crash can't leave a seen-row without its count. Put the same event through a hundred times and you get one. The consumer commits its Kafka offset only after the database transaction commits (enable.auto.commit=false, commit by hand), so the only failure mode left is "see an event again", which is now harmless.

events_seen grows at the rate of the topic. I keep thirty days, partitioned by week on day rather than seen_at — the unique key has to contain the partition column, and day is the one thing a replayed event can't change (see the partitioning note) — and drop the oldest partition nightly. Dedup tables that nobody prunes are a thing people find eighteen months later, usually while looking for disk.

Step 3: deploy the idempotent consumer before replaying anything, and note the group's committed offsets at that moment. Everything after those offsets is handled by code that can be replayed safely. Everything before them is the awkward middle: events the old consumer counted, with no seen-row to prove it.

The awkward middle

There are two honest ways through it. The general one is to rebuild the affected days from scratch: pause the other writer, zero the days in the window, and run the whole window through the new consumer under a fresh group id. Every event lands once because events_seen starts empty for that range. That's what I'd do today if the second writer could be paused, and it's the path any future replay will take.

It couldn't be paused, so I did the targeted version: a one-off consumer, with its own group id, that reads the window and processes only the events the bug dropped — the ones with a negative amount — through the same idempotent statement. Those were never counted, so counting them now is correct, and the seen-row protects against the one-off being run twice.

from confluent_kafka import Consumer, TopicPartition

c = Consumer({
    "bootstrap.servers": BROKERS,
    "group.id": "counts-backfill-2026-07",   # never the production group
    "enable.auto.commit": False,
    "auto.offset.reset": "error",
})

# seek every partition to the first offset at or after START_MS
parts = [TopicPartition(TOPIC, p, START_MS) for p in range(NUM_PARTITIONS)]
c.assign(c.offsets_for_times(parts))

done = set()
while len(done) < NUM_PARTITIONS:
    msg = c.poll(1.0)
    if msg is None:
        continue
    p = msg.partition()
    if msg.offset() >= stop_offsets[p]:     # production group's offset at deploy time
        done.add(p)
        continue
    ev = decode(msg.value())
    if ev["amount"] >= 0:
        continue                             # only what the bug dropped
    with db.transaction():
        db.execute(UPSERT_SQL, ev["event_id"], ev["day"], ev["customer_id"])

stop_offsets came from kafka-consumer-groups --describe on the production group, run right after the fixed consumer went live. Anything at or past those offsets was seen by the new code and has a seen-row; replaying it would be a no-op anyway, but stopping there keeps the run bounded.

Two things to do before you start

Check retention. The topic had seven days of retention and the window started six days back. I bumped it to fourteen first — kafka-configs --alter --add-config retention.ms=1209600000 — which takes effect without a restart. Don't backfill from a topic that might delete the segments underneath you halfway through.

Decide how you'll know it worked. I snapshotted daily_counts for the window before the run. Afterwards, the per-day difference had to equal the number of negative-amount events per day that the one-off logged. It did, to the row, which is the only reason I slept that night.

The general shape

Land with a unique key. Derive the aggregate from what you landed, not from what you received. Commit the offset after the database, never before. Once those three hold, a replay is just a consumer with a different group id, and a backfill is a Tuesday afternoon instead of a week. The increment was the bug; the refund filter just found it.