Add field to rd_kafka_queue which denotes if it contains fetched msgs

[milindl]

An issue in v2.1.0 was fixed in which max.poll.interval.ms was not honored,
because it was reset on any queue poll, not just consumer poll.
It was changed it so that only certain rdkafka.h functions which were polling
would reset the timer.

However, librdkafka exposes a method rd_kafka_queue_get_consumer, which returns
the consumer queue, and the application can poll this queue for events rather
than calling consume poll. There is no way to distinguish polls to this queue
and an arbitrary queue, and it won't reset the timer.

So, a new field is maintained inside the queue denoting if it might
contain fetched messages, or not. It deals with forwarding of queues, so
if a queue which receives fetched messages is forwarded multiple times,
calling poll on the forwardee will also reset the timer.

Note that there are a few points where someone might forward/unforward into a queue while polling it. (See cases with does_q_contain_consumer_msgs = rd_kafka_q_consumer_cnt(rkq, 1);) which might make the timer reset wrong on these occasions.

I don't think it matters much since that's a somewhat convoluted use case and to fix it will require taking a lock over the duration of the entire poll. But let me know what you think.

Fixes confluentinc/confluent-kafka-go#980

[pranavrth]

As we are locking on polling, please check performance once.

[emasab]

Great Milind, it was difficult to find the correct combination of changes that solves both problems by resetting the timer only when needed and fits with the existing code.

[milindl]

Thanks Emanuele. Will merge after #4262 for CI

[pranavrth]

LGTM!