Add ConsumerListener

[ollis]

I would appreciate the possibility of being able to use the underlying consumer for configuration of, for example, metrics.
Something similar to this: https://github.com/reactor/reactor-kafka/blob/main/src/main/java/reactor/kafka/receiver/MicrometerConsumerListener.java

[nomisRev]

Hey @ollis,

Thanks out for the feature request 🙏 I am thinking about how I can design this in a better way, because it also needs to expose the underlying Consumer being used in a stream such that user triggered pause can be tracked inside the loop. I will take this into account as well 👍