Kafka Input operator - needs better handling of failure of Kafka reader thread

Kafka input operator has a separate thread to read from Kafka. Even if the Kafka reader thread is not running, the operator will continue to run.

Ideally, following things needs to be done

1. More logging as to why the Kafka reader thread exited
2. If the Kafka reader thread gets killed either kill the container or retry with a new thread.

Note: We have seen an intermittent issue, where Kafka Consumer thread has exited, but operator continues to be reported as in good health because the exception is not thrown in the operator thread. After making this change, I have seen this issue only once and also Kafka logged the message along the lines (fetch failed)