my-skills

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

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

Deduplication (inbox) with EF Core

  1. Dedup table with a UNIQUE key on a stable business id (header/payload — not offset); durable, not an in-memory HashSet.
  2. 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).
  3. Bounded window — purge/TTL the dedup table; it must not grow forever.

Idempotent effects

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

Producer

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

Outbox, retry & rebalance

  1. 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.
  2. 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

  1. 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).