kafka-idempotency

name

kafka-idempotency

description

Safe-idempotency review for consuming and producing Kafka messages in .NET / ASP.NET Core (Confluent.Kafka). Use when reviewing consumers (BackgroundService/IHostedService), producers, and handlers — checks per-message DI scope, DbContext lifetime, deduplication (inbox) with EF Core, the correct persist-then-commit-offset order (EnableAutoCommit=false / StoreOffset), singleton IProducer, delivery via ProduceAsync/delivery report, Flush and graceful shutdown (Close), the Outbox pattern against dual-write, retry/DLQ, and rebalance.

Kafka + .NET — Idempotency in Consuming and Producing

Safe idempotency for Kafka consumers/producers in .NET / ASP.NET Core (Confluent.Kafka). Focus on the .NET code (BackgroundService, DI, DbContext, EF Core), not broker theory. Act as a senior .NET engineer: direct, technical, no praise. Review only the changed lines of the diff.

Premise

Kafka is at-least-once — the same message will be redelivered (rebalance, crash before offset commit, replay). Idempotency is the handler's responsibility, not the broker's.

The safe recipe

Order is the guarantee: effect + dedup mark in one DB transaction → commit DB → only then commit/store the offset. Crash anywhere → redelivery → dedup absorbs it.

protected override async Task ExecuteAsync(CancellationToken ct)
{
    await Task.Yield();                 // don't block host startup (Consume is blocking)
    _consumer.Subscribe("orders");
    try
    {
        while (!ct.IsCancellationRequested)
        {
            var r = _consumer.Consume(ct);
            if (r?.Message is null) continue;
            using var scope = _scopeFactory.CreateScope();    // 1 scope per message
            var handler = scope.ServiceProvider.GetRequiredService<IOrderHandler>();
            await handler.HandleAsync(r.Message, ct);         // effect + dedup mark (1 tx)
            _consumer.StoreOffset(r);                         // ONLY after the DB commit
        }
    }
    catch (OperationCanceledException) { }
    finally { _consumer.Close(); }      // final commit + clean group exit
}
// handler: _db.ProcessedMessages.Add(new(messageId)); ApplyEffect(_db, msg); await SaveChangesAsync();
// catch DbUpdateException when unique-violation => already processed, skip (effect not reapplied).

Consumer & DI

1. Loop in a BackgroundService, never a controller. Consume is blocking → await Task.Yield() at the top; never async void.
2. One DI scope per message (IServiceScopeFactory.CreateScope()). Never inject DbContext/other Scoped into the singleton consumer — it becomes a de-facto singleton, not thread-safe, serves stale data.
3. A handler exception must not kill the loop — catch and route to retry/DLQ. Never advance the offset on failure (= silent loss).

Offset & commit

4. EnableAutoCommit = false (or EnableAutoOffsetStore = false + StoreOffset after processing). Never let a timer advance the offset independently of processing.
5. Persist before committing the offset. Offset and DB aren't atomic together → dedup covers the gap. Commit/StoreOffset store offset + 1.
6. No naive parallelism (Task.Run per message) — breaks per-partition order and can commit N+1 before N finishes.

Deduplication (inbox) with EF Core

7. Dedup table with a UNIQUE key on a stable business id (header/payload — not offset); durable, not an in-memory HashSet.
8. Mark + effect in one SaveChanges. Treat the unique violation as "already processed" — the DB constraint is the real guard (check-then-insert races between instances).
9. Bounded window — purge/TTL the dedup table; it must not grow forever.

Idempotent effects

10. Prefer upsert (MERGE/ON CONFLICT); use absolute values, not increments (balance = 100, never += 10).
11. Idempotency-Key on external calls (API/payment/email) so the downstream deduplicates.
12. Non-transactional effects after the commit (HTTP, email) — never inside a block that can roll back.

Producer

13. IProducer<,> as a reused singleton — thread-safe and expensive; new ...Build() per message is a bug (exhausts connections, loses batching).
14. EnableIdempotence = true (forces acks=all, max.in.flight<=5). It doesn't dedup across restarts → use Outbox for a guaranteed publish.
15. Confirm delivery — await ProduceAsync or handle the delivery report; fire-and-forget = silent loss. Flush/dispose on shutdown.

Outbox, retry & rebalance

16. No dual-write. SaveChanges + ProduceAsync in separate steps isn't atomic — write the event to an outbox table in the same tx; a relay (polling/CDC) publishes with retry. The consumer still deduplicates.
17. Bounded retry + DLQ for poison messages; handle SetPartitionsRevokedHandler; keep per-message work within max.poll.interval.ms (or the broker triggers a rebalance + redelivery).

Config, observability & tests

18. Producer EnableIdempotence=true+Acks=All; consumer EnableAutoCommit=false, IsolationLevel=ReadCommitted for transactional topics. Log messageId/correlationId. Test double-delivery (single effect) and the crash between SaveChanges and the offset commit.

Kafka transactions give exactly-once only Kafka→Kafka. .NET consumers writing to DB/HTTP still need app-level dedup.

Output format

Ignore what is correct — list only what needs to change, ordered by impact:

#	Severity	Category	Comment	Rationale
1	🔴 Critical	DI / DbContext	DbContext injected into the BackgroundService constructor	De-facto singleton, not thread-safe, stale data; use a per-message scope
2	🔴 Critical	Offset	StoreOffset called before SaveChanges	Crash between the two loses the message (at-most-once)
3	🟠 High	Dual-write	SaveChanges() then ProduceAsync() separately	Not atomic; a failed publish leaves state without an event — use Outbox
…	…	…	…	…

Severities: 🔴 Critical (duplicate effect, message loss, mismanaged DbContext, dual-write), 🟠 High (config/offset/producer/fragile dedup), 🟡 Medium (best practice), 🔵 Low (style).