Don't reset max.poll.interval.ms limit on every poll

[milindl]

Earlier on, we changed the code to prevent max.poll.interval.ms from being triggered in case we were inside librdkafka in any sort of a poll call. Top achieve this, blocked the timer using rd_kafka_app_poll_blocking, and reset it at the end of the call.

This doesn't work correctly in the cases where we're simply polling an unrelated queue, like the log queue. In that case, just by polling the log queue, the timer is reset, despite us not actually consuming anything (or doing any consume poll).

At the same time, it's a reasonable expectation that max.poll.interval.ms won't be triggered while we are doing any sort of consumer poll.

This PR takes care of both the cases. The methods (from public API) which block/reset the timer are:

1. rd_kafka_consume_batch
2. rd_kafka_consume_batch_queue
3. rd_kafka_consume_callback
4. rd_kafka_consume_callback_queue
5. rd_kafka_consume
6. rd_kafka_consumer_poll

All the other poll methods like queue_poll etc no longer reset or block the timer.
A failing test was added first, to test the change.

Ref #2365.

[pranavrth]

I would suggest using different methods for different tests.

1. It improves readability
2. We are not touching the old test case just adding a new one

[milindl]

Done, also added a changelog entry

[emasab]

I think if we're adding some new checks to an old test case, it's ok to change it as long as we don't change existing asserts.
This is also to avoid having tests that are too slow, especially integration tests.
It can also be ok with to parameterize some tests to avoid code duplication, as long as the tests are very similar and the resulting code is readable.

What do you think @pranavrth?
That said @milindl you don't have to revert this change you can leave them separate.

[milindl]

I am okay with either option as per what we decide.
Both these tests are slow though and can't be run in make quick because of the timeout.

[pranavrth]

I felt that we are touching multiple places in the old test so asked to separate.
Though anything is fine as far as the old test is not changed.

[emasab]

LGTM

[vbmithr]

Just want to mention that this "fix" broke my binding where I use events and read consumer events from the consumer queue (and never calling rd_kafka_consumer_poll). I solve the problem with polling rd_kafka_consumer_poll but it would be more elegant if polling from the consumer queue would reset the timeout as well.

[milindl]

Yep, we're aware of the bug already. We're looking into it with #4256 . Thanks.