Allow committed offsets refresh

[rtimush]

This pull request solves the problem described in https://issues.apache.org/jira/browse/KAFKA-3806 and #301.

In some workloads, it is possible that no events are sent to a given topic/partition for 24 hours. By default, Kafka evicts committed offsets after this period, so if the consumer is then restarted it has no stored offset to start with. Depending on the auto.offset.reset it may result in re-processing of the whole history, losing some recent events or a failure.

This pull request allows a consumer to refresh the committed offset on a regular basis even if there are no new messages in the topic. The following logic applies:

• If the last commit has been successfully completed, the offsets from this commit are sent.
• If there were no messages at all since the consumer started, the original offset for a partition is sent.
• If the last commit has failed, no offset refresh for the partitions affected happens.

The feature should be explicitly enabled by calling withCommitRefreshInterval on the ConsumerSettings, so existing client won't be affected.

There is an integration test for this feature, but it takes a few minutes to run because the eviction timeout is configured in minutes and cannot be fractional.

[bsideup]

@rtimush @ktoso @patriknw do you have any updates on the status of this PR?

[fchaillou]

I worked with the reporter of #301 and i can confirm that this would perfectly fix our issue.
Hope to see it coming in a future version of reactive-kafka

[ktoso]

We'll soon have @ennru join us and spend considerable time on Alpakka projects, which includes this, which means quicker turnaround in merging things very soon :-) Thanks for your patience, Alpakka has proven it self and is going to get the attention it deserves very soon :-)

[fchaillou]

@ktoso no worries, I'm interested in these features so i'm doing what i can to help you in my own way !
Welcome and good luck to @ennru then 👍

[ennru]

Good work and welcome to the Kafka connector project!

Could you rebase, please?

[rtimush]

Thanks, @ennru. Rebased.

[ennru]

Great, thank you.
I'd like to see some documentation for this feature. We'd at least need a section in reference.conf.

[rtimush]

Makes sense, I added parameters to the reference.conf and also added a small note in the consumer docs.

[ennru]

LGTM

[ktoso]

would it not be nicer to have one value, and it being infinite being "disabled"? That's usually how we have such settings in Akka and Akka HTTP

[rtimush]

It is indeed much nicer, but I didn't want to introduce a custom config behavior. As you say it is already done like this in akka, I will change the code here to use a single config value too.

[ktoso]

One comment to simplify settings, looks good otherwise

[ennru]

LGTM, better with just one setting.

[ennru]

Thank you for your contribution!