Wrap Kafka Consummer onMessage with exception handling.

* Make KafkaConsumerConnector resilient to commit exceptions #1043

* Code review rework

* Code review
1 file changed