Periodic committer does not commit already committed offsets

[bsideup]

In Kafka, the offsets are stored for a limited period of time (24h by default), and if the consumer did not commit them during that period, they get wiped.
There is "periodic commit" logic in reactor-kafka, but it removes consumedOffsets during the commit:

reactor-kafka/src/main/java/reactor/kafka/receiver/internals/CommittableBatch.java

Line 67 in 3bd1962

iterator.remove();

If partition did not receive any events for the past 24h, no commit will be performed and Kafka will delete the offset for that partition.

Reactor Kafka should support a "refresh" of already committed offsets to avoid that.

See a similar issue in Akka's reactive-kafka:
akka/alpakka-kafka#375