When a fetch response receives an error about partition leadership, fencing, etc. a metadata refresh is triggered. However, it takes time for that refresh to occur, and in the interim, it appears that the consumer will blindly attempt to fetch data for the partition again, in kind of a "definition of insanity" type of way. Ideally, the consumer would have a way to temporarily ignore those partitions, in a way somewhat like the "pausing" approach so that they are skipped until the metadata refresh response is fully processed.
This affects both the existing KafkaConsumer and the new PrototypeAsyncConsumer.