Remove the kafka_stop call on error, as "errors" come in for recoverable situations like a broker in a pool not being able to be connected to.  Calling kafka_stop() causes memory to be freed downstream which then causes segfaults