KAFKA-2654: optimize unnecessary poll(0) away

[ymatsuda]

@guozhangwang
This change aims to remove unnecessary consumer.poll(0) calls.

• once after some partition is resumed
• whenever the size of the top queue in any task is below BUFFERED_RECORDS_PER_PARTITION_CONFIG

[guozhangwang]

I feel the second condition is still too restrictive: BUFFERED_RECORDS_PER_PARTITION_CONFIG is used to pause a certain partition and hence is usually large, probably we can define another config for a lower-threshold?

Also could you change the PR title with this Kafka ticket?

https://issues.apache.org/jira/browse/KAFKA-2654

I think it is not a minor issue and hence better be crediting as a major contribution.

[ymatsuda]

The second condition checks only the top queue. It means that, if the top queue has records more than BUFFERED_RECORDS_PER_PARTITION_CONFIG, we may not poll records even if there are queues with sizes below BUFFERED_RECORDS_PER_PARTITION_CONFIG. This mechanism is orthogonal to the secondary threshold you are talking about.

[guozhangwang]

I see, that is right. The second condition is actually similar to the first one but just "pre-fetch" with one iteration earlier. LGTM.