KIP-822 — Optimize the semantics of KafkaConsumer#pause to be consistent between the two RebalanceProtocols

Discussion Consumer

Fixes a bug where `KafkaConsumer.pause()` state is incorrectly cleared during rebalance in both EAGER and COOPERATIVE protocols, causing previously paused partitions to resume delivering records unexpectedly. `ConsumerCoordinator.invokePartitionsRevoked()` clears the paused set without flushing corresponding buffered fetch data, leading to duplicate or unintended record delivery after the pause is cleared.

Details