fix(consumer): prevent offset loss during rebalance under high load

[nimanikoo]

Description

This PR fixes a race condition in KafkaFlow where consumers were losing track of offsets during Kafka partition rebalances under high load, causing messages to be processed more than once.

Root cause: During a rebalance (OnPartitionsRevoked / OnPartitionsAssigned), the WorkerPoolFeeder was not stopped before the ConsumerWorkerPool was torn down and rebuilt. This caused in-flight EnqueueAsync calls to unblock into the new OffsetManager, which had no knowledge of the revoked partition — resulting in silent no-ops, untracked offsets, and duplicate message processing after the rebalance.

Why StopAsync() couldn't be called directly: Rebalance callbacks are invoked synchronously from within consumer.Consume() on the same thread the feeder is blocked on. Awaiting the feeder task there would cause a deadlock.

Fix — two complementary layers:

• Layer 1: Added IWorkerPoolFeeder.Cancel(), which signals cancellation without awaiting the feeder task. Both rebalance handlers now call Cancel() before stopping the pool and Start() after the pool is ready.
• Layer 2: OffsetManager.Enqueue() now returns bool and rejects messages for untracked partitions. ConsumerWorkerPool.EnqueueAsync() discards such messages instead of forwarding them to a worker without offset tracking. Kafka will replay skipped messages on the new partition owner after the rebalance completes.

Fixes # (issue)

How Has This Been Tested?

5 new unit tests were added:

Test	Covers
ConsumerManagerCooperativeStickyTests.OnPartitionsRevoked_ShouldCancelFeederBeforeWorkerPool_*	Cooperative sticky rebalance: revoke path
ConsumerManagerCooperativeStickyTests.OnPartitionsAssigned_ShouldCancelAndRestartFeeder_*	Cooperative sticky rebalance: assign path
ConsumerManagerTests.OnPartitionsRevoked_ShouldCancelFeederBeforeWorkerPool_*	Stop-the-world (RoundRobin): revoke path
ConsumerManagerTests.OnPartitionsAssigned_ShouldCancelAndRestartFeeder_*	Stop-the-world: assign path
OffsetManagerTests.Enqueue_ForRevokedPartition_SilentlyDropsMessage_*	Layer 2 safety net behavior

Each rebalance test also verifies call order (Cancel → StopPool → StartPool → Start), not just method invocation. All 124 unit tests pass.

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

Disclaimer

By sending us your contributions, you are agreeing that your contribution is made subject to the terms of our Contributor Ownership Statement

[nimanikoo]

Thanks again for approving my Kafka Flow PR. 🙌 @bblankenship78

Looks like the repository now requires 2 approving reviews before a PR can be merged. Would you mind helping me get one more approval? If there's someone you think could review it quickly, I'd really appreciate it. Thanks!

[therealpaulgg]

ran into some issues that I think this PR would solve. Can we get this merged into kafkaflow?