Use offset_fetch api_version when fetching messages

[GDegrove]

When switching to the new kayrock client, we noticed that the api_version for the auto_commit of the fetch message is not taken into account.
Therefore, even if using the API version 2 for the offset commit with kayrock, the next fetch will always use the api_version 0

This introduces a divergence in the way the new client can be used compared to the old implementation. As it's up to the user code to provide the correct offset to fetch.

[sourcelevel-bot]

Hello, @GDegrove! This is your first Pull Request that will be reviewed by SourceLevel, an automatic Code Review service. It will leave comments on this diff with potential issues and style violations found in the code as you push new commits. You can also see all the issues found on this Pull Request on its review page. Please check our documentation for more information.

[jbruggem]

Apart from my comment, this looks good to me. Given that it's a sensitive piece of code and that I worked on this with you, I'd rather we had a second approval.

@dantswain @joshuawscott could either of you take a look ? It's a short one :)

[jbruggem]

Nit-pick on this: I'm wondering if we shouldn't offer a choice to the user whether they wish to use offset_commit_api_version also for offset_fetch_api_version.

In any case, I'm thinking that maybe some explicit variable name would help the reader follow the logic and make it explicit that we've decided that offset_fetch_api_version = offset_commit_api_version.

Suggestion:

    offset_commit_api_version = Keyword.get(opts, :offset_commit_api_version, 0)
    # By default, it makes sense to synchronize the API version of the offset_commit and the offset_fetch
    # operations, otherwise we might commit the offsets in zookeeper and read them from Kafka, meaning
    # that the value would be incorrect.    
    offset_fetch_api_version = Keyword.get(opts, :offset_fetch_api_version, offset_commit_api_version)

    retrieved_offset =
      current_offset(
        supplied_offset,
        partition,
        topic,
        worker_name,
        offset_fetch_api_version,

[GDegrove]

It makes real sense, I pushed the correction :)

[jbruggem]

OK for me. Once we have a second approval, we can merge.

[bjhaid]

lgtm

[jbruggem]

Thanks a lot @bjhaid !