KAFKA-19259: Async consumer fetch intermittent delays on console consumer

[kirktrue]

There’s a difference in the two consumers’ pollForFetches() methods in this case: ClassicKafkaConsumer doesn't block waiting for data in the fetch buffer, but AsyncKafkaConsumer does.

In ClassicKafkaConsumer.pollForFetches(), after enqueuing the FETCH request, the consumer makes a call to ConsumerNetworkClient.poll(). In most cases poll() returns almost immediately because it successfully sent the FETCH request. So even when the pollTimeout value is, e.g. 3000, the call to ConsumerNetworkClient.poll() doesn't block that long waiting for a response.

After sending out a FETCH request, AsyncKafkaConsumer then calls FetchBuffer.awaitNotEmpty() and proceeds to block there for the full length of the timeout. In some cases, the response to the FETCH comes back with no results, which doesn't unblock FetchBuffer.awaitNotEmpty(). So because the application thread is still waiting for data in the buffer, it remains blocked, preventing any more FETCH requests from being sent, causing the long pauses in the console consumer.