Actually expose ability to set rebalance listener

[ktoso]

Seems this was attempted in 2016 in #93 however something was not quite done "all the way" it seems.

Always the hardcoded

private[kafka] def rebalanceListener(onAssign: Set[TopicPartition] => Unit, onRevoke: Set[TopicPartition] => Unit): ListenerCallbacks =
    ListenerCallbacks(onAssign, onRevoke)

    def rebalanceListener =
      KafkaConsumerActor.rebalanceListener(tps => partitionAssignedCB.invoke(tps), partitionRevokedCB.invoke)

is used instead of one provided by the user.

It seems like it was intended to be able to pass it in a Subscription, but was never done?

[ktoso]

cc @kciesielski @patriknw

I'll work on this as got someone requesting it