Synchronous partition assignment handler (internal)

[ennru]

Purpose

For now this PR just introduces the synchronous partition assignment handler internally without a user-facing API as the use-cases for this need to be understood better.

Introduce a new callback for changes in partitions assigned to an Alpakka Kafka consumer. This allows executing code before a rebalance continues, and querying the underlying Kafka consumer.

These callbacks are much more powerful (and dangerous) than the current rebalance listener actor. The callbacks are executed on the same thread that called poll() (the actor thread) so they are allowed to call into the Kafka consumer eg. to query current offsets, commit offsets or to seek to an offset position. The callbacks add to the time poll() is executed and block the internal Kafka consumer actor from doing any other work. As this can easily make influence other Kafka consumer actors running on the same dispatcher, the time the callbacks take is checked and warnings are issued to the logs if they pass a configurable threshold.

This allows for new ways to handle offsets which might show more efficient for certain use-cases. The tests show an example for external offset storage and one example which commits on rebalance only.

Background Context

It has been discussed in #539 that the lack of blocking user-defined call-backs to react on Kafka's partition rebalancing makes certain use cases impossible with Alpakka Kafka.

In particular, it is impossible to combine Kafka's partition assignment with external offset storage.

References

References #539
Follow up #841

[seglo]

This looks really promising @ennru . When I was considering an API I was thinking about providing strategies that users could opt in with (i.e. "commit on revoked partitions"), and then provide a super user interface which just exposed the consumer itself, but I like the idea behind RestrictedConsumer better.

[2m]

Looking good. I like how TransactionalSource was able to use this without any code changes.

[ennru]

For now, I removed the user-facing API so that the new partition assignment handler is used internally, only.
The plan is to continue to explore the scenarios this kind of partition assignment handler improves as eg. #841 and #843

[2m]

LGTM