Store cluster client in Broker struct for reusability

Signed-off-by: chrismark <chrismarkou92@gmail.com>

metricbeat/module/kafka/broker.go

@@ -43,6 +43,7 @@ func Version(version string) kafka.Version {
 type Broker struct {
 	broker *sarama.Broker
 	cfg    *sarama.Config
+	client sarama.Client
 
 	advertisedAddr string
 	id             int32
@@ -96,6 +97,7 @@ func NewBroker(host string, settings BrokerSettings) *Broker {
 	return &Broker{
 		broker:  sarama.NewBroker(host),
 		cfg:     cfg,
+		client:  nil,
 		id:      noID,
 		matchID: settings.MatchID,
 	}
@@ -104,6 +106,7 @@ func NewBroker(host string, settings BrokerSettings) *Broker {
 // Close the broker connection
 func (b *Broker) Close() error {
 	closeBroker(b.broker)
+	b.client.Close()
 	return nil
 }
 
@@ -134,6 +137,13 @@ func (b *Broker) Connect() error {
 	debugf("found matching broker %v with id %v", other.Addr(), other.ID())
 	b.id = other.ID()
 	b.advertisedAddr = other.Addr()
+
+	c, err := getClusteWideClient(b.Addr(), b.cfg)
+	if err != nil {
+		closeBroker(b.broker)
+		return fmt.Errorf("Could not get cluster client for advertised broker with address %v", b.Addr())
+	}
+	b.client = c
 	return nil
 }
 
@@ -272,15 +282,10 @@ func (b *Broker) FetchGroupOffsets(group string, partitions map[string][]int32)
 
 // GetPartitionOffsetFromTheLeader fetches the OffsetNewest from the leader.
 func (b *Broker) GetPartitionOffsetFromTheLeader(topic string, partitionID int32) (int64, error) {
-	client, err := sarama.NewClient([]string{b.Addr()}, b.cfg)
-	if err != nil {
-		return -1, err
-	}
-	offset, err := client.GetOffset(topic, partitionID, sarama.OffsetNewest)
+	offset, err := b.client.GetOffset(topic, partitionID, sarama.OffsetNewest)
 	if err != nil {
 		return -1, err
 	}
-	defer client.Close()
 	return offset, nil
 }
 
@@ -538,6 +543,14 @@ func anyIPsMatch(as, bs []net.IP) bool {
 	return false
 }
 
+func getClusteWideClient(addr string, cfg *sarama.Config) (sarama.Client, error) {
+	client, err := sarama.NewClient([]string{addr}, cfg)
+	if err != nil {
+		return nil, err
+	}
+	return client, nil
+}
+
 func brokerAddresses(brokers []*sarama.Broker) []string {
 	addresses := make([]string, len(brokers))
 	for i, b := range brokers {