[Bug] Kafka: message committed without executing subscriber during shutdown, then replayed out of order after FallbackWindowLookbackSeconds

Summary

Kafka: message committed without executing subscriber during shutdown, then replayed out of order after FallbackWindowLookbackSeconds

Steps to Reproduce

During a rolling restart, a received message can be committed to Kafka without its subscriber ever running. The row stays in Scheduled state and is only picked up by the retry processor after FallbackWindowLookbackSeconds. The result is a message on a single partition executing minutes after a higher-offset message from the same partition, with Retries = 0 and no warning or error in the logs.

I understand CAP is at-least-once and does not promise partition ordering after a subscriber failure. This report is about a case where no failure happened and no subscriber code ran: the dispatcher refused the message because CAP was stopping, the caller could not see that, and it committed the offset anyway. The shutdown path that should have prevented this waits on a task that is always already completed.

Environment

• DotNetCore.CAP 10.0.2 (code references below are against master at e52b8508)
• Transport: Kafka (Confluent.Kafka 2.14.2), enable.auto.commit = false (CAP default)
• Storage: PostgreSQL
• FallbackWindowLookbackSeconds = 360, EnableSubscriberParallelExecute = false, no GroupConcurrent
• Kubernetes, several replicas, rolling restart

Observed behaviour

Two messages M1 (lower offset) and M2 (higher offset) on the same partition:

M1 executed 362 s after it was stored: FallbackWindowLookbackSeconds plus the wait for the next retry tick (up to FailedRetryInterval). Its row had Added ≈ produce time and status Scheduled until then.

Root cause

1. On host shutdown Bootstrapper cancels its CTS (IBootstrapper.Default.cs#L122). That token is linked into both the Dispatcher (IDispatcher.Default.cs#L60) and the ConsumerRegister (IConsumerRegister.Default.cs#L59), so the dispatcher stops accepting work at the same instant polling is asked to stop. There is no ordering between the two.

2. ConsumerRegister.Dispose waits on _compositeTask (#L95), but that field is always Task.CompletedTask (#L165). The listening loops are fire-and-forget (#L134), so nothing is actually drained.

3. The Kafka loop only checks the token at the top of each iteration (KafkaConsumerClient.cs#L107) and Consume(timeout) (#L113) does not take it, so one more message can be returned after cancellation. The callback stores it as Scheduled (IDataStorage.PostgreSql.cs#L197) and calls EnqueueToExecute.

4. EnqueueToExecute sees cancellation and returns silently, without logging (IDispatcher.Default.cs#L118-L121). If it gets past that check, SubscribeExecutor throws OperationCanceledException before invoking the subscriber (ISubscribeExector.Default.cs#L90) and the dispatcher swallows it (#L132-L135). The method returns a plain ValueTask, so the caller cannot distinguish "executed" from "dropped".

5. The very next line commits the offset (IConsumerRegister.Default.cs#L260). Kafka now considers the message done and never redelivers it.

6. KafkaConsumerClient.DisposeAsync calls Dispose() without Close() (KafkaConsumerClient.cs#L157-L161). Per the Confluent.Kafka docs for Consumer.Dispose, no leave-group request is sent and the group only rebalances after session.timeout.ms (default 45 s). With the default eager assignment strategy that stalls the whole group. M2's ~3 minute delay is consistent with several pods doing this in sequence during the rollout, though I have not proven that part from CAP code alone.

7. The Scheduled row is later returned by the retry query, which selects Scheduled as well as Failed rows older than the lookback (IDataStorage.PostgreSql.cs#L349-L350), and executed by MessageNeedToRetryProcessor (IProcessor.NeedRetry.cs#L97-L103). Retries is only incremented on the failure path (ISubscribeExector.Default.cs#L144), so the audit trail looks like a clean first attempt.

Ordering is silently handed from Kafka's offset-ordered redelivery to CAP's time-based fallback, which knows nothing about partitions.

Expected behaviour

1. No commit when the dispatcher refuses. If EnqueueToExecute declines a message because CAP is stopping, ConsumerRegister must not commit the offset. Leaving it uncommitted (or calling RejectAsync) lets the next assignee receive it from the broker in order.
2. Ordered shutdown. Stop polling first, wait for in-flight callbacks with a bounded timeout, then cancel the dispatcher. Today _compositeTask does not track the listening loops, so the existing 2 s wait is a no-op.
3. A dropped message must be visible. The early return in EnqueueToExecute should log at warning level like its EnqueueToPublish counterpart does.
4. Kafka: Close() before Dispose() so the group rebalances immediately instead of after session.timeout.ms.

Possible fix directions

• Have EnqueueToExecute return a ValueTask<bool> (accepted or refused). In ConsumerRegister, commit only on true; on false skip the commit (and optionally delete the just-stored row so the fallback does not execute it out of order later).
• Track the listening tasks in _compositeTask and cancel the consumer token before the dispatcher token, with a configurable drain timeout.
• In KafkaConsumerClient.DisposeAsync, call _consumerClient.Close() before Dispose().

Expected Behavior

No response

Actual Behavior

No response

Log Output

CAP Configuration

Transport Used

None

Storage Provider

None

Environment

No response

Additional Context

No response