Merge pull request #47 from amuraru/broker_log_dirs

Fix broker describe log dirs

Author: weeco

kafka/cluster.go

@@ -149,6 +149,9 @@ func (module *Cluster) describeLogDirs() {
 
 	for _, broker := range brokers {
 		go func(b *sarama.Broker) {
+			// try to open the connection with broker if not already established
+			// ignoring the error as the only possible one at this stage is AlreadyConnected
+			_ = b.Open(module.client.Config())
 			res, err := b.DescribeLogDirs(req)
 			resCh <- response{
 				BrokerID: b.ID(),