Fix exception handling for reactive consumers

[giamo]

Version 4.1.0 introduced a bug for reactive consumers, the DefaultKafkaListenerExceptionHandler is always used even when a bean implements its own KafkaListenerExceptionHandler.

The problem was introduced here #441, there are a couple of calls to the handleException() that expects a consumerBean which are receiving a consumerState. Added a test to verify that the simple fix works.

cc @dstepanov

[CLAassistant]

All committers have signed the CLA.

[dstepanov]

Thanks!