Remove availableBroker func

We can get the correct broker by selecting the controller

kafka/client.go

@@ -302,21 +302,3 @@ func (c *Client) topicConfig(topic string) (map[string]*string, error) {
 	}
 	return conf, nil
 }
-
-func (c *Client) availableBroker() (*sarama.Broker, error) {
-	var err error
-	brokers := *c.config.BootstrapServers
-	kc := c.kafkaConfig
-
-	log.Printf("[DEBUG] Looking for Brokers @ %v", brokers)
-	for _, b := range brokers {
-		broker := sarama.NewBroker(b)
-		err = broker.Open(kc)
-		if err == nil {
-			return broker, nil
-		}
-		log.Printf("[WARN] Broker @ %s cannot be reached\n", b)
-	}
-
-	return nil, fmt.Errorf("No Available Brokers @ %v", brokers)
-}

kafka/kafka_acls.go

@@ -90,10 +90,11 @@ func tfToAclFilter(s stringlyTypedACL) (sarama.AclFilter, error) {
 }
 
 func (c *Client) DeleteACL(s stringlyTypedACL) error {
-	broker, err := c.availableBroker()
+	broker, err := c.client.Controller()
 	if err != nil {
 		return err
 	}
+
 	filter, err := tfToAclFilter(s)
 	if err != nil {
 		return err
@@ -118,7 +119,7 @@ func (c *Client) DeleteACL(s stringlyTypedACL) error {
 }
 
 func (c *Client) CreateACL(s stringlyTypedACL) error {
-	broker, err := c.availableBroker()
+	broker, err := c.client.Controller()
 	if err != nil {
 		return err
 	}
@@ -210,7 +211,7 @@ func stringToAclPermissionType(in string) sarama.AclPermissionType {
 }
 
 func (c *Client) ListACLs() ([]*sarama.ResourceAcls, error) {
-	broker, err := c.availableBroker()
+	broker, err := c.client.Controller()
 	if err != nil {
 		return nil, err
 	}