Fix offset lost

[mvSapphire]

Description

Messages are silently skipped when the worker pool is restarted by a workers count change (e.g. WithConsumerLagWorkerBalancer). A partition can permanently lose an offset — offset 2 is never processed while 3, 4, … are consumed and committed — and because the commit moves past the gap, a process restart does not recover it either.

The root cause is that the OffsetManager lifetime is tied to the worker pool instead of to the partition assignment. ConsumerWorkerPool.StopAsync sets _offsetManager = null and StartAsync builds a fresh one, but a workers count change does not rewind the librdkafka read position. Everything the discarded OffsetManager still had pending is forgotten, and the new instance happily commits over it.

Two distinct paths lead to the loss:

1. Pending contexts are dropped with the OffsetManager

WaitContextsCompletionAsync() is awaited with .WithCancellation(token, false), so it returns silently once WorkerStopTimeout elapses. Contexts that were never MarkAsProcessed (cancelled mid-processing, or not completed when AutoMessageCompletion is disabled) stay in PartitionOffsets._receivedContexts and are thrown away with the manager. The new manager starts empty, so the next completed message becomes the head of the queue and its offset is committed — writing straight over the gap.

2. Messages are dropped before reaching the OffsetManager

ChangeWorkersCountAsync stops the feeder first, which cancels the token passed to IWorkerDistributionStrategy.GetWorkerAsync. BytesSumDistributionStrategy (the default) and PartitionKeyDistributionStrategy return null on that token, and FreeWorkerDistributionStrategy throws OperationCanceledException from Channel.Reader.ReadAsync, which the feeder loop swallows. In both cases ConsumerWorkerPool.EnqueueAsync returned without enqueuing, so a message that was already read from Kafka is never tracked anywhere. This path does not require WorkerStopTimeout to expire at all — with FreeWorkerDistributionStrategy the window is "any moment all workers are busy".

Changes

• IConsumerWorkerPool.StopAsync takes a keepOffsetManager flag.
• ConsumerWorkerPool.StartAsync only creates an OffsetManager (and starts the OffsetCommitter) when the pool does not already own one, so the offset bookkeeping now follows the partition assignment rather than the workers.
• ConsumerWorkerPool.StopAsync keeps the OffsetManager and the OffsetCommitter alive when asked to.
• ConsumerWorkerPool.EnqueueAsync no longer drops a message that was already read from Kafka. The OperationCanceledException is handled, the context is registered in the OffsetManager and then discarded, so it blocks the commit instead of disappearing.
• ConsumerManager.ChangeWorkersCountAsync stops the pool with keepOffsetManager: true. Partition revoke/assign and consumer shutdown keep the previous behaviour, which is correct there because Kafka re-reads from the committed offset after a rebalance.

This also removes a related inconsistency: contexts capture the OffsetManager they were created with, so a detached task from the previous pool generation finishing late used to mark offsets on a discarded manager that still shared the live OffsetCommitter, pushing a stale offset into the commit. With a single manager across the restart, late completions land in the right place.

Behaviour change

An unprocessed message no longer disappears — instead the committed offset for that partition freezes at the gap and the message is delivered again the next time the consumer starts from the committed offset (process restart or a rebalance). This is the expected at-least-once behaviour and it is already what a discarded message does within a single OffsetManager lifetime; the change only makes it survive a worker pool restart. The visible consequences are that broker-side lag for the affected partition stops draining until then, and messages after the gap are redelivered.

Fully removing the freeze would require seeking the partition back to the first unprocessed offset on restart. The internal IConsumer abstraction does not expose Seek today, so that is left out of this PR.

Fixes # (issue)

How Has This Been Tested?

New ConsumerWorkerPoolOffsetTrackingTests drives a real ConsumerWorkerPool with a real OffsetManager and OffsetCommitter, captures the committed offsets and covers one test per loss path plus a happy path:

• ChangingWorkersCount_WithMessageNotProcessed_ShouldNotCommitPastIt — fails on master with Expected CommittedOffsets() to contain only items matching (offset <= 1), but {4L} do(es) not match.
• ChangingWorkersCount_WithMessageNotAssignedToAnyWorker_ShouldNotCommitPastIt — fails on master with {3L} do(es) not match.
• ChangingWorkersCount_WithAllMessagesProcessed_ShouldCommitTheLastOffset — passes before and after, guarding against over-correcting and stalling commits when nothing is pending.

ConsumerManagerTests.ChangingWorkersCount_StopsWorkerPoolKeepingTheOffsetManager covers the wiring and fails if ChangeWorkersCountAsync goes back to StopAsync().

The existing StopAsync mock setups in ConsumerManagerTests and ConsumerManagerCooperativeStickyTests were made explicit (StopAsync(false)) because Moq expression trees cannot use optional arguments. They keep asserting the previous behaviour for the revoke/assign/shutdown paths.

Full unit test suite: 123 passed, 0 failed. Integration tests were not run (they require a live broker).

Checklist

• My code follows the style guidelines of this project
• I have performed a self-review of my own code
• I have added tests to cover my changes
• I have made corresponding changes to the documentation

[mvSapphire]

@brmagadutra @joelfoliveira Hi guys, I’d really appreciate it if you could take a look.