Three duplicates, three causes, one symptom

Three bugs in one week

In the last week of August I looked at duplicated numbers in three different systems. Two of them live in the pipeline that takes playback logs from screens to ClickHouse through Kafka. The third one is in an e-commerce API that sends paid orders to a sales system. Different teams, different languages, code from different years. I only saw that they were the same problem when I wrote the three side by side.

1. Two consumers that were never in the group

Our import job is simple. A cron starts a Python worker, the worker reads a batch from Kafka and inserts it in ClickHouse. With subscribe() the consumer joins its group, the group coordinator splits the partitions, and two workers each get a half. With assign() you pick the partitions yourself and the coordinator is out of the picture. The worker still commits offsets using the group id, but it's never a member of the group.

from confluent_kafka import Consumer, TopicPartition

c = Consumer({
    "bootstrap.servers": BROKERS,
    "group.id": "audit-import",
})

# what the worker did: take all 6 partitions, start from the committed offset
c.assign([TopicPartition("playlog", p) for p in range(6)])

# what two workers need: let the coordinator split the partitions
c.subscribe(["playlog"])

With one worker this is harmless, and that's what makes it dangerous. When a second worker starts, both take all six partitions from the same committed offset and every message goes in twice. The only dedup in the worker was a Python set() inside the batch, which can't see the other process, so the factor at insert was exactly 2.0. The crontab even tried to start them at different times with a */2+1 and */2+2 step. That syntax is not valid in the cron we run, and it doesn't fail: the offset part is ignored and both workers fired in the same second, as the syslog showed.

If you look, the group itself tells you: the committed offsets keep moving every 2 minutes and the group has 0 members.

$ kafka-consumer-groups.sh --bootstrap-server $BROKERS --describe --group audit-import --state

GROUP          COORDINATOR (ID)   ASSIGNMENT-STRATEGY   STATE   #MEMBERS
audit-import   broker-2 (2)                             Empty   0

Empty, no members and no assignment strategy, while the lag stays small: that's how assign() looks from the broker side when you describe the group.

There was a second thing in the same function. The worker called consumer.close() in a finally before the insert, and with auto-commit that writes the offset first. If the insert fails, the batch is gone. Committing after the write is the order I'd use in any consumer, even if a failed insert means the same batch comes again and the dedup has to handle it:

# consumer created with "enable.auto.commit": False
msgs = c.consume(num_messages=30_000, timeout=10)
rows = [parse(m) for m in msgs if not m.error()]
clickhouse.insert("playlog", rows)      # if this raises, nothing is committed
c.commit(asynchronous=False)            # offsets only after the data is written

2. A summary that read before the merge

In the raw table, a ReplacingMergeTree, duplicates with the same sorting key collapse when parts merge, so the raw table heals itself, eventually. On top of it a summarizer runs every 10 minutes, counts rows per device, campaign and day, and inserts into a SummingMergeTree. Summing only adds. Whatever double count goes in there stays, because a SummingMergeTree has no idea that two of its inputs were the same event.

How eventual is eventual? I measured on this table in August, after a period where every row was written 2 times:

Our summarizer ran right in the 10 minute line. It used count() over the raw table without FINAL, so it was counting physical rows. To measure this on your own table, compare rows with distinct keys for a recent window of 10 minutes:

SELECT
    count()                          AS rows,
    uniqExact(device_id, event_id)   AS keys,
    round(rows / keys, 4)            AS dup_factor
FROM playlog
WHERE inserted_at >= now() - INTERVAL 10 MINUTE;

I didn't want to believe my own number, so it went through three independent analyses, one of them with the explicit task to refute it. Two checks settled it. The control group: another environment runs the same code with one worker and its factor was 1.0000. The outside source: the sum in the summary table matched the rows ClickHouse says it inserted, from system.part_log, at a ratio of 0.99994 for a whole month. So the summary was counting rows written, and from the day the second worker started the rows written were double.

SELECT sum(rows)
FROM system.part_log
WHERE event_type = 'NewPart'
  AND database = 'audit' AND table = 'playlog'
  AND event_date BETWEEN '2026-01-01' AND '2026-01-31';

For the summary, any of these works: read with FINAL (it costs CPU, it's correct), summarize only windows older than the time your merges take, or group by the dedup key before counting. The first one is the simplest to review and the one I'd start with, because the cost shows up in the query log and you can measure it:

INSERT INTO playlog_summary
SELECT device_id, campaign_id, toDate(ts) AS day, count() AS total
FROM playlog FINAL
WHERE inserted_at >= {from:DateTime} AND inserted_at < {to:DateTime}
GROUP BY device_id, campaign_id, day;

Removing the second worker stopped the bleeding the same day, and we re-summarized the past months from the raw history. One detail I almost missed: without the second worker the factor came back to about 1.10, not 1.00. There was an older and smaller source of duplicates, a separate story. If I had declared victory at the first number, the older problem would still be in the summary today without anyone looking for it.

3. Two webhooks for one payment

A payment gateway sends order.paid and charge.paid almost at the same moment. Both handlers did the reasonable thing, and enqueued "integrate order X". The sales system received the same order twice. The fix in BullMQ 5 is one option on add():

await this.queue.add(
  "integrate-order",
  { orderId },
  { deduplication: { id: `order-${orderId}` } },
);

My first version had ttl: 60_000. In the review of the same day I took the TTL out. Without it, BullMQ keeps the key while there's a job waiting or running for that order and frees it when the job finishes, so a manual re-integration later still works and there is no magic number for how far apart two webhooks can arrive. This only protects the enqueue. The receiving side should still treat the order id as an idempotency key, because a retry after a timeout can arrive from anywhere. (This was the same week of the stuck sync I wrote about in query_timeout doesn't cancel anything.)

The same shape

Duplicated processing is almost never the same bug twice. It's almost always the same missing piece: a key that says "this already happened", and every layer checking it. In each case some layer was trusting another one to do the dedup. The raw table had dedup, eventually. The worker had a set(), per process, and the queue had nothing at all until that week in August.

None of these were careless mistakes. Each piece was reasonable alone, and bugs like the assign() one only wake up when you scale out, which is usually when there is a backlog and nobody is looking at consumer groups. The question I ask now for every layer is what happens if it runs twice, and the cheapest test is to run it twice on purpose in staging and count rows against distinct keys.

Filed under the problem it solves

• deduplication
• idempotency
• kafka consumer
• clickhouse
• webhooks
• bullmq
• reliability