Description

In case of subscription change with a consumer using the cooperative assignor it can resume fetching from a previous position.
That can also happen if resuming a partition that wasn't paused.
The reason is that fetch version is immediately bumped when enqueuing the resume operation and even if the operation is then discarded because partition isn't paused, the is no next_fetch_start to start from when offset validation is completed in rd_kafka_toppar_fetch_decide_start_from_next_fetch_start, or it's the next_fetch_start of previous pause operation, so there can be a reset to an offset that was already consumed.

How to reproduce

Execute test 0050/test_no_duplicate_messages("cooperative-sticky") or 0145/test_no_duplicate_messages_unnecessary_resume(*) in #4636.

Checklist

Please provide the following information:

• librdkafka version (1.x)
• Apache Kafka version: <REPLACE with e.g., 0.10.2.3>
• librdkafka client configuration: <REPLACE with e.g., message.timeout.ms=123, auto.reset.offset=earliest, ..>
• Operating system: <REPLACE with e.g., Centos 5 (x64)>
• Provide logs (with debug=.. as necessary) from librdkafka
• Provide broker log excerpts
• Critical issue