Fix cluster-wide Kafka client bug (#14857) (#14862)

(cherry picked from commit 6c279eb)
elastic · Nov 29, 2019 · a269249 · a269249
1 parent 311bab6
commit a269249
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 7 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
 }
 

diff --git a/metricbeat/module/kafka/consumergroup/consumergroup_integration_test.go b/metricbeat/module/kafka/consumergroup/consumergroup_integration_test.go
@@ -81,7 +81,7 @@ func TestFetch(t *testing.T) {
 	for retries := 0; retries < 3; retries++ {
 		data, errors = mbtest.ReportingFetchV2Error(f)
 		if len(data) > 0 {
-			break
+			continue
 		}
 		time.Sleep(500 * time.Millisecond)
 	}