Kafka consumer must commit offsets asynchronously

The offset commit calls to Kafka may occasionally take very long.
In that case, the notifyCheckpointComplete() method blocks for long and the KafkaConsumer cannot make progress and cannot perform checkpoints.

Kafka 0.9+ have methods to commit asynchronously.
We should use those and make sure no more than one commit is concurrently in progress, to that commit requests do not pile up.