Drop already buffered messages of revoked partitions

[ennru]

Short description

When a partition is revoked from an Alpakka Kafka consumer, it continues to emit data which was buffered before. This can be up to max.poll.records messages that got fetched just before the revoke was issued.

Instead, the stage should drop messages from its buffer when it doesn't have the partition assigned anymore.

Fixing this could reduce the number of re-delivered messages at rebalances significantly.

Details

After receiving messages the KafkaConsumerActor.processResult passes them to the stages that requested messages for that topic/partition. The stage does not reconsider messages in its buffer when the Kafka broker rebalances a consumer group, and will continue to issue messages that belong to partitions it no longer "owns" to the stream.

[ennru]

This is similar to what the TransactionalSource does with its InflightRecords.

[ennru]

Fixed with #946 and #992