Repository: Shopify/sarama
Pull request #1645 (2020-03-19 03:38:51)

Remove broker(s) which no longer exist in metadata
This pull request solves the problem: #1629. I find it only have update and add broker when update brokers by latest metadata, however there is no code of remove broker which if offline

Stephan14: Hey @d1egoaz, I have found some error as follows:
![image](https://user-images.githubusercontent.com/13467996/77030980-3d92b300-69db-11ea-916d-eaed8122f45b.png)
 And run test on my Mac, it show as follows:
![image](https://user-images.githubusercontent.com/13467996/77031019-5d29db80-69db-11ea-9fe6-cb755a2913b5.png)
I think `Got err: kafka: client has run out of available brokers to talk to` is only one log and the code is right, so I delete this log
Stephan14: Hey @bai @d1egoaz,   I find other PR appear some error on CI, for example https://github.com/Shopify/sarama/pull/1640/checks?check_run_id=504088005,  but it check successful. So how to solve the problem.

d1egoaz: Thanks @Stephan14 for the new PR, so it seems the changes that you are introducing are indeed changing/breaking some expectation in some tests. We'll need to 👀 what's going on.
Stephan14: Hey @d1egoaz , how about this problem is going? Can I merge this PR to master? The failures is as follow:
![image](https://user-images.githubusercontent.com/13467996/77449766-59f47c80-6e2d-11ea-9b3b-bde73f3bf754.png)

Stephan14: @dnwe , oh, sorry. I have fixed the problem and run `go test` on my mac successfully as follows:
![image](https://user-images.githubusercontent.com/13467996/78672105-c3d35280-7912-11ea-9789-bef73e7ea475.png)
But it runs failed in some CI as follows:
![image](https://user-images.githubusercontent.com/13467996/78672275-01d07680-7913-11ea-8607-bf4a28ff01e7.png)
How to fix it

Patch:
diff --git a/client.go b/client.go
index 81a3ca5..c3392f9 100644
--- a/client.go
+++ b/client.go
@@ -562,6 +562,30 @@ func (client *client) RefreshCoordinator(consumerGroup string) error {
 
 // private broker management helpers
 
+func (client *client) updateBroker(brokers []*Broker) {
+	var currentBroker = make(map[int32]*Broker, len(brokers))
+
+	for _, broker := range brokers {
+		currentBroker[broker.ID()] = broker
+		if client.brokers[broker.ID()] == nil { // add new broker
+			client.brokers[broker.ID()] = broker
+			Logger.Printf("client/brokers registered new broker #%d at %s", broker.ID(), broker.Addr())
+		} else if broker.Addr() != client.brokers[broker.ID()].Addr() { // replace broker with new address
+			safeAsyncClose(client.brokers[broker.ID()])
+			client.brokers[broker.ID()] = broker
+			Logger.Printf("client/brokers replaced registered broker #%d with %s", broker.ID(), broker.Addr())
+		}
+	}
+
+	for id, broker := range client.brokers {
+		if _, exist := currentBroker[id]; !exist { // remove old broker
+			safeAsyncClose(broker)
+			delete(client.brokers, id)
+			Logger.Printf("client/broker remove invalid broker #%d with %s", broker.ID(), broker.Addr())
+		}
+	}
+}
+
 // registerBroker makes sure a broker received by a Metadata or Coordinator request is registered
 // in the brokers map. It returns the broker that is registered, which may be the provided broker,
 // or a previously registered Broker instance. You must hold the write lock before calling this function.
@@ -885,10 +909,9 @@ func (client *client) updateMetadata(data *MetadataResponse, allKnownMetaData bo
 	// For all the brokers we received:
 	// - if it is a new ID, save it
 	// - if it is an existing ID, but the address we have is stale, discard the old one and save it
+	// - if some brokers is not exist in it, remove old broker
 	// - otherwise ignore it, replacing our existing one would just bounce the connection
-	for _, broker := range data.Brokers {
-		client.registerBroker(broker)
-	}
+	client.updateBroker(data.Brokers)
 
 	client.controllerID = data.ControllerID
 
diff --git a/config.go b/config.go
index e899820..08d50ce 100644
--- a/config.go
+++ b/config.go
@@ -96,9 +96,8 @@ type Config struct {
 			GSSAPI GSSAPIConfig
 		}
 
-		// KeepAlive specifies the keep-alive period for an active network connection (defaults to 0).
-		// If zero or positive, keep-alives are enabled.
-		// If negative, keep-alives are disabled.
+		// KeepAlive specifies the keep-alive period for an active network connection.
+		// If zero, keep-alives are disabled. (default is 0: disabled).
 		KeepAlive time.Duration
 
 		// LocalAddr is the local address to use when dialing an
@@ -538,6 +537,8 @@ func (c *Config) Validate() error {
 		return ConfigurationError("Net.ReadTimeout must be > 0")
 	case c.Net.WriteTimeout <= 0:
 		return ConfigurationError("Net.WriteTimeout must be > 0")
+	case c.Net.KeepAlive < 0:
+		return ConfigurationError("Net.KeepAlive must be >= 0")
 	case c.Net.SASL.Enable:
 		if c.Net.SASL.Mechanism == "" {
 			c.Net.SASL.Mechanism = SASLTypePlaintext
diff --git a/consumer_group.go b/consumer_group.go
index 056b9e3..0bc32b9 100644
--- a/consumer_group.go
+++ b/consumer_group.go
@@ -760,7 +760,7 @@ func (s *consumerGroupSession) heartbeatLoop() {
 		case ErrRebalanceInProgress, ErrUnknownMemberId, ErrIllegalGeneration:
 			return
 		default:
-			s.parent.handleError(resp.Err, "", -1)
+			s.parent.handleError(err, "", -1)
 			return
 		}
 
diff --git a/list_partition_reassignments_response.go b/list_partition_reassignments_response.go
index 4baa6a0..a5786ee 100644
--- a/list_partition_reassignments_response.go
+++ b/list_partition_reassignments_response.go
@@ -1,19 +1,19 @@
 package sarama
 
 type PartitionReplicaReassignmentsStatus struct {
-	Replicas         []int32
-	AddingReplicas   []int32
-	RemovingReplicas []int32
+	replicas         []int32
+	addingReplicas   []int32
+	removingReplicas []int32
 }
 
 func (b *PartitionReplicaReassignmentsStatus) encode(pe packetEncoder) error {
-	if err := pe.putCompactInt32Array(b.Replicas); err != nil {
+	if err := pe.putCompactInt32Array(b.replicas); err != nil {
 		return err
 	}
-	if err := pe.putCompactInt32Array(b.AddingReplicas); err != nil {
+	if err := pe.putCompactInt32Array(b.addingReplicas); err != nil {
 		return err
 	}
-	if err := pe.putCompactInt32Array(b.RemovingReplicas); err != nil {
+	if err := pe.putCompactInt32Array(b.removingReplicas); err != nil {
 		return err
 	}
 
@@ -23,15 +23,15 @@ func (b *PartitionReplicaReassignmentsStatus) encode(pe packetEncoder) error {
 }
 
 func (b *PartitionReplicaReassignmentsStatus) decode(pd packetDecoder) (err error) {
-	if b.Replicas, err = pd.getCompactInt32Array(); err != nil {
+	if b.replicas, err = pd.getCompactInt32Array(); err != nil {
 		return err
 	}
 
-	if b.AddingReplicas, err = pd.getCompactInt32Array(); err != nil {
+	if b.addingReplicas, err = pd.getCompactInt32Array(); err != nil {
 		return err
 	}
 
-	if b.RemovingReplicas, err = pd.getCompactInt32Array(); err != nil {
+	if b.removingReplicas, err = pd.getCompactInt32Array(); err != nil {
 		return err
 	}
 
@@ -60,7 +60,7 @@ func (r *ListPartitionReassignmentsResponse) AddBlock(topic string, partition in
 		r.TopicStatus[topic] = partitions
 	}
 
-	partitions[partition] = &PartitionReplicaReassignmentsStatus{Replicas: replicas, AddingReplicas: addingReplicas, RemovingReplicas: removingReplicas}
+	partitions[partition] = &PartitionReplicaReassignmentsStatus{replicas: replicas, addingReplicas: addingReplicas, removingReplicas: removingReplicas}
 }
 
 func (r *ListPartitionReassignmentsResponse) encode(pe packetEncoder) error {
@@ -110,7 +110,7 @@ func (r *ListPartitionReassignmentsResponse) decode(pd packetDecoder, version in
 	}
 
 	numTopics, err := pd.getCompactArrayLength()
-	if err != nil {
+	if err != nil || numTopics == 0 {
 		return err
 	}
 


Test patch:
diff --git a/client_test.go b/client_test.go
index 2beef6c..cd47a50 100644
--- a/client_test.go
+++ b/client_test.go
@@ -441,6 +441,8 @@ func TestClientReceivingPartialMetadata(t *testing.T) {
 	replicas := []int32{leader.BrokerID(), seedBroker.BrokerID()}
 
 	metadataPartial := new(MetadataResponse)
+	metadataPartial.AddBroker(seedBroker.Addr(), 1)
+	metadataPartial.AddBroker(leader.Addr(), 5)
 	metadataPartial.AddTopic("new_topic", ErrLeaderNotAvailable)
 	metadataPartial.AddTopicPartition("new_topic", 0, leader.BrokerID(), replicas, replicas, []int32{}, ErrNoError)
 	metadataPartial.AddTopicPartition("new_topic", 1, -1, replicas, []int32{}, []int32{}, ErrLeaderNotAvailable)
@@ -485,6 +487,7 @@ func TestClientRefreshBehaviour(t *testing.T) {
 	seedBroker.Returns(metadataResponse1)
 
 	metadataResponse2 := new(MetadataResponse)
+	metadataResponse2.AddBroker(leader.Addr(), leader.BrokerID())
 	metadataResponse2.AddTopicPartition("my_topic", 0xb, leader.BrokerID(), nil, nil, nil, ErrNoError)
 	seedBroker.Returns(metadataResponse2)
 
@@ -512,6 +515,36 @@ func TestClientRefreshBehaviour(t *testing.T) {
 	safeClose(t, client)
 }
 
+func TestClientRefreshMetadataBrokerOffline(t *testing.T) {
+	seedBroker := NewMockBroker(t, 1)
+	leader := NewMockBroker(t, 5)
+
+	metadataResponse1 := new(MetadataResponse)
+	metadataResponse1.AddBroker(leader.Addr(), leader.BrokerID())
+	metadataResponse1.AddBroker(seedBroker.Addr(), seedBroker.BrokerID())
+	seedBroker.Returns(metadataResponse1)
+
+	client, err := NewClient([]string{seedBroker.Addr()}, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if len(client.Brokers()) != 2 {
+		t.Error("Meta broker is not 2")
+	}
+
+	metadataResponse2 := new(MetadataResponse)
+	metadataResponse2.AddBroker(leader.Addr(), leader.BrokerID())
+	seedBroker.Returns(metadataResponse2)
+
+	if err := client.RefreshMetadata(); err != nil {
+		t.Error(err)
+	}
+	if len(client.Brokers()) != 1 {
+		t.Error("Meta broker is not 1")
+	}
+}
+
 func TestClientResurrectDeadSeeds(t *testing.T) {
 	initialSeed := NewMockBroker(t, 0)
 	emptyMetadata := new(MetadataResponse)
@@ -656,7 +689,6 @@ func TestClientMetadataTimeout(t *testing.T) {
 
 			// Start refreshing metadata in the background
 			errChan := make(chan error)
-			start := time.Now()
 			go func() {
 				errChan <- c.RefreshMetadata()
 			}()
@@ -666,7 +698,6 @@ func TestClientMetadataTimeout(t *testing.T) {
 			maxRefreshDuration := 2 * timeout
 			select {
 			case err := <-errChan:
-				t.Logf("Got err: %v after waiting for: %v", err, time.Since(start))
 				if err == nil {
 					t.Fatal("Expected failed RefreshMetadata, got nil")
 				}
diff --git a/config_test.go b/config_test.go
index 6da9dbe..a7267b1 100644
--- a/config_test.go
+++ b/config_test.go
@@ -67,6 +67,11 @@ func TestNetConfigValidates(t *testing.T) {
 				cfg.Net.WriteTimeout = 0
 			},
 			"Net.WriteTimeout must be > 0"},
+		{"KeepAlive",
+			func(cfg *Config) {
+				cfg.Net.KeepAlive = -1
+			},
+			"Net.KeepAlive must be >= 0"},
 		{"SASL.User",
 			func(cfg *Config) {
 				cfg.Net.SASL.Enable = true
diff --git a/consumer_test.go b/consumer_test.go
index 7b29ebc..20ee8dd 100644
--- a/consumer_test.go
+++ b/consumer_test.go
@@ -640,6 +640,7 @@ func TestConsumerRebalancingMultiplePartitions(t *testing.T) {
 		"MetadataRequest": NewMockMetadataResponse(t).
 			SetBroker(leader0.Addr(), leader0.BrokerID()).
 			SetBroker(leader1.Addr(), leader1.BrokerID()).
+			SetBroker(seedBroker.Addr(), seedBroker.BrokerID()).
 			SetLeader("my_topic", 0, leader0.BrokerID()).
 			SetLeader("my_topic", 1, leader1.BrokerID()),
 	})
@@ -720,7 +721,10 @@ func TestConsumerRebalancingMultiplePartitions(t *testing.T) {
 	seedBroker.SetHandlerByMap(map[string]MockResponse{
 		"MetadataRequest": NewMockMetadataResponse(t).
 			SetLeader("my_topic", 0, leader1.BrokerID()).
-			SetLeader("my_topic", 1, leader1.BrokerID()),
+			SetLeader("my_topic", 1, leader1.BrokerID()).
+			SetBroker(leader0.Addr(), leader0.BrokerID()).
+			SetBroker(leader1.Addr(), leader1.BrokerID()).
+			SetBroker(seedBroker.Addr(), seedBroker.BrokerID()),
 	})
 
 	// leader0 says no longer leader of partition 0
@@ -759,7 +763,10 @@ func TestConsumerRebalancingMultiplePartitions(t *testing.T) {
 	seedBroker.SetHandlerByMap(map[string]MockResponse{
 		"MetadataRequest": NewMockMetadataResponse(t).
 			SetLeader("my_topic", 0, leader1.BrokerID()).
-			SetLeader("my_topic", 1, leader0.BrokerID()),
+			SetLeader("my_topic", 1, leader0.BrokerID()).
+			SetBroker(leader0.Addr(), leader0.BrokerID()).
+			SetBroker(leader1.Addr(), leader1.BrokerID()).
+			SetBroker(seedBroker.Addr(), seedBroker.BrokerID()),
 	})
 
 	// leader1 provides three more messages on partition0, says no longer leader of partition1