Messages are silently lost when a handler throws and enable.auto.commit is true

[nabisobhi]

Describe the bug

When enable.auto.commit is true — the default in ConsumerConfigurationBase — a message whose handler throws is marked as consumed and is never redelivered. The exception aborts processing, but the offset has already been stored by the Kafka client and gets committed anyway, so the message is silently lost.

The loss is deterministic on a graceful stop, not merely a timing race. Consumer.ConsumeAll holds the consumer scope in a using:

https://github.com/dfds/dafda/blob/master/src/Dafda/Consuming/Consumer.cs

public async Task ConsumeAll(CancellationToken cancellationToken)
{
    using var consumerScope = consumerScopeFactory.CreateConsumerScope();
    while (!cancellationToken.IsCancellationRequested)
    {
        await ProcessNextMessage(consumerScope, cancellationToken);
    }
}

When a handler exception propagates out of ProcessNextMessage, that using disposes the scope during stack unwinding — before ConsumerHostedService's catch block runs. KafkaConsumerScope.Dispose() calls _innerKafkaConsumer.Close(), and Close() commits stored offsets when auto-commit is enabled. The offset itself was stored the moment Consume() returned the message, since Dafda never sets enable.auto.offset.store and librdkafka defaults it to true.

Net effect: the commit lands before the configured IConsumerErrorHandler has decided anything, so both ConsumerFailureStrategy.Default (stop the app) and RestartConsumer skip the message.

ProcessNextMessage only issues an explicit commit in manual mode, so nothing counteracts this:

if (!isAutoCommitEnabled)
{
    await messageResult.Commit(cancellationToken);
}

To Reproduce

1. Configure a consumer with defaults, i.e. without setting enable.auto.commit (it defaults to true).
2. Register a message handler that throws for a given message.
3. Produce that message and let the consumer pick it up.
4. Let the application stop and restart (the default failure strategy stops it).
5. The message is not redelivered — the consumer group's committed offset has advanced past it.

Expected behavior

A message that was not processed successfully should not have its offset committed, so it is redelivered when the consumer resumes — the at-least-once guarantee that manual commit mode already provides.

At minimum, the failure path should not commit an offset for a message that never completed.

Additional context

Scope. This is long-standing behavior and predates the dead letter queue work — the if (!isAutoCommitEnabled) guard around the explicit commit, with no exception handling around dispatch, was present before #351. It is reachable today with stock configuration and no dead letter queue configured.

Not a duplicate of #16. #16 asks for a new periodic-manual-commit mode as a throughput optimization, explicitly accepting re-delivery as its trade-off, and mentions auto-commit message loss only as background framing. It neither tracks this as a defect nor proposes anything that would fix the auto-commit failure path. This issue is about correctness of the existing auto-commit mode.

Interaction with the dead letter queue bypass (#353). The bypass added in #353 lets chosen exception types skip the dead letter queue so the service fails fast instead of draining a topic into the DLQ during a systemic outage. Under default configuration that intent is defeated: the crash commits the offset of the very message it crashed on. #353 documents the limitation on the public API rather than changing default behavior, which is why this is filed separately.

Possible direction. Set enable.auto.offset.store=false and explicitly store the offset after successful handling. That keeps auto-commit's batching and throughput while ensuring only successfully handled messages have their offsets advanced, and it is the standard librdkafka approach for at-least-once with auto-commit. Whether this becomes the default or an opt-in mode needs a decision, since it changes delivery semantics for existing auto-commit users.