Fix cluster-wide Kafka client bug

Signed-off-by: chrismark <[email protected]>
elastic · Nov 29, 2019 · 4f0b4f9 · 4f0b4f9
1 parent 8859ead
commit 4f0b4f9
Showing 1 changed file with 7 additions and 6 deletions.
diff --git a/metricbeat/module/kafka/broker.go b/metricbeat/module/kafka/broker.go
@@ -116,6 +116,13 @@ func (b *Broker) Connect() error {
 		return errors.Wrap(err, "broker.Open failed")
 	}
 
+	c, err := getClusterWideClient(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
+
 	if b.id != noID || !b.matchID {
 		return nil
 	}
@@ -138,12 +145,6 @@ func (b *Broker) Connect() error {
 	b.id = other.ID()
 	b.advertisedAddr = other.Addr()
 
-	c, err := getClusterWideClient(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
 }