error handling when kafka not ready

6 years ago · ab85118233
1 changed files with 3 additions and 0 deletions
--- a/weed/notification/kafka/kafka_queue.go
+++ b/weed/notification/kafka/kafka_queue.go
@ -37,6 +37,9 @@ func (k *KafkaQueue) initialize(hosts []string, topic string) (err error) {
 	config.Producer.Return.Successes = true
 	config.Producer.Return.Errors = true
 	k.producer, err = sarama.NewAsyncProducer(hosts, config)
+	if err != nil {
+		return err
+	}
 	k.topic = topic
 	go k.handleSuccess()
 	go k.handleError()