fix Consumer to emit error when it cannot connect to Kafka