Kafka: commit-strategy overhaul — replace per-message Commit() with CommitMode (StoreThenAutoFlush default) #3150

Child of #3134 (Re-Evaluate Kafka Integration). This is the foundational, immediate first step of the Kafka re-evaluation and the original problem reported in #3134.

Problem

KafkaListener.CompleteAsync calls _consumer.Commit() with no arguments on every message (KafkaListener.cs:92):

public ValueTask CompleteAsync(Envelope envelope)
{
    try { _consumer.Commit(); } catch (Exception) { }
    return ValueTask.CompletedTask;
}

Two problems:

1. Throughput — Commit() is a synchronous broker round-trip executed once per message. This is the ~20x bottleneck reported in Re-Evaluate Kafka Integration #3134.
2. Semantics — argument-less Commit() commits the consumer's current position across all assigned partitions, not the specific envelope's offset. It's only (mostly) safe today because consumption is effectively serial; it would silently over-commit the moment any buffering/parallelism is added.

EnableAutoCommit is set to false in Durable mode (in KafkaTopic.BuildListenerAsync / KafkaTopicGroup.BuildListenerAsync), so we already do all commit work by hand — in the least efficient way possible. StoreOffset / EnableAutoOffsetStore are never used.

Goal

Replace the hard-coded per-message commit with an explicit, opt-in commit strategy, defaulting to the idiomatic non-blocking Kafka model.

Scope

1. Respect EnableAutoCommit

If a user enables auto-commit, stop issuing manual commits entirely.

2. CommitMode (working name) on KafkaListenerConfiguration

• PerMessage — current behavior, kept for strict at-least-once with minimal reprocessing.
• StoreThenAutoFlush (proposed default) — set EnableAutoOffsetStore = false, call StoreOffset(result) after successful processing, and let Kafka's background committer flush on AutoCommitIntervalMs. Non-blocking, at-least-once, idiomatic high-throughput. (This is exactly the model requested in Re-Evaluate Kafka Integration #3134.)
• BatchCount / BatchInterval — Wolverine-managed async Commit after N messages or T elapsed (Spring AckMode / Alpakka Committer style), with an offset-reordering guard so we never commit ahead of the lowest in-flight offset (the contiguous-offset watermark / "OffsetManager").

3. Commit the specific offset (correctness fix)

CompleteAsync must StoreOffset(tpo) / Commit(tpo) for the envelope's own TopicPartitionOffset, never the global position. Required before any concurrency work. (Envelope already carries Offset and PartitionId, set in KafkaListener.cs:48-49.)

4. Graceful-shutdown flush

Flush stored/pending offsets in StopAsync so a clean shutdown doesn't lose progress.

Relationships

• Provides the offset-watermark gate that the buffered path of Kafka: opt-in intra-partition concurrency by key (durable, reuse ShardedExecutionBlock) #3140 (intra-partition concurrency by key) depends on.
• @mbar10 offered a PR in Re-Evaluate Kafka Integration #3134 — this issue is where the StoreThenAutoFlush work fits; coordinate the config surface here.
• The DeferAsync blocking-retry fix is handled in Kafka: non-blocking retry topics (tiered fixed-delay) + MoveToKafkaRetryTopic action #3148 (retry topics), but interacts with commit semantics — keep consistent.

Background refs

• Confluent: StoreOffset + EnableAutoOffsetStore=false + background auto-commit is the recommended high-throughput model — .NET client overview
• Historical gotcha to validate against: confluent-kafka-dotnet#1635 (EnableAutoOffsetStore=false still auto-committing in some versions).

Acceptance criteria

• Manual commits suppressed when EnableAutoCommit = true.
• CommitMode config with PerMessage, StoreThenAutoFlush (default), BatchCount, BatchInterval.
• CompleteAsync commits/stores the envelope's specific TopicPartitionOffset, not the global position.
• Batch modes never commit ahead of the lowest in-flight offset (contiguous watermark).
• Offsets flushed on graceful shutdown.
• Tests: throughput model issues no per-message synchronous commit; offsets advance correctly; crash/restart resumes from the last flushed offset without loss; specific-offset commit verified under (simulated) out-of-order completion.

Out of scope

• Intra-partition concurrency by key (Kafka: opt-in intra-partition concurrency by key (durable, reuse ShardedExecutionBlock) #3140) — consumes the watermark gate built here.