Committing: explore different implementation

[ennru]

Committing: explore different implementation

The current Alpakka Kafka implementation for committing offsets to Kafka relies on collecting offsets for processed messages in the stream. In case of a change in the consumer group, Kafka revokes and assigns partitions which implies that the consumer can't commit any further offsets to partitions that are revoked (and not re-assigned). This means already in-stream collected offsets can't be committed and those messages will be re-emitted to the consumer that gets that partition assigned by Kafka.

Committing on partition revoke

The experiment of introducing a PartitionAssignmentHandler in #761 explores how such a handler could be used to implement committing outside of the current Alpakka Kafka handling. A key idea is to leverage the callback from Kafka when partitions are revoked to commit any offsets which are known to have been processed successfully.

If committing can be handled just before a partition gets revoked, much fewer messages will get reprocessed during consumer rebalancing.

Potential improvements

1. Fewer messages get reprocessed on rebalance
2. It would become possible to allow for some out-of-order processing if the committing logic keeps track of holes in the offset list and just commits segments that are done.

Potential negative changes

1. More actor messaging as every message sends its commit to the collector
2. Larger change in the Alpakka Kafka architecture

Necessary changes

1. New "Commit collector" actor as child of KafkaConsumerActor
2. Change commitScaladsl to send the offset to the "Commit collector" (fire and forget?)
3. Schedule regular commits from "Commit collector" (using committer max-interval and max-batch)
4. Commit any uncommitted offsets from the partition revoke by asking the "Commit collector"
5. Remove batching from the "Committer"

[ennru]

#862 and #868 implement the aggregation of commits.
I'll create a new issue to discuss dropping of commit back-pressure.