fix(client): add nilguards to updateBroker (#3393)

This _should_ be a rare edge case as the updateBroker func is only
called from updateMetadata which does already do an up-front
`client.Close()` check under read-lock before then acquiring the write
lock. However, there's potentially a small window of opportunity that if
client.Close() was called whilst metadata refresh was in-flight and for
whatever reason the updateMetadata goroutine gets pre-empted in-between
the readlock release and the write lock acquire then the client could
have been closed and so `client.brokers` will be nil.

I wouldn't have expected this to ever happen, but it was reported by a
user in an older Sarama version in #3391 and there's no harm in adding
the nilguard just in case.

Signed-off-by: Dominic Evans <[email protected]>

Author: dnwe

client.go

@@ -685,9 +685,16 @@ func (client *client) randomizeSeedBrokers(addrs []string) {
 }
 
 func (client *client) updateBroker(brokers []*Broker) {
+	if client.brokers == nil {
+		return
+	}
+
 	currentBroker := make(map[int32]*Broker, len(brokers))
 
 	for _, broker := range brokers {
+		if broker == nil {
+			continue
+		}
 		currentBroker[broker.ID()] = broker
 		if client.brokers[broker.ID()] == nil { // add new broker
 			client.brokers[broker.ID()] = broker

client_test.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/rcrowley/go-metrics"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestSimpleClient(t *testing.T) {
@@ -1330,3 +1331,70 @@ func TestMetricsCleanup(t *testing.T) {
 		t.Errorf("excepted 1 metric, found: %v", all)
 	}
 }
+
+func TestUpdateBroker(t *testing.T) {
+	t.Run("closed client doesn't panic", func(t *testing.T) {
+		c := &client{}
+		fn := func() {
+			c.updateBroker(nil)
+			c.updateBroker([]*Broker{
+				{
+					id:   0,
+					addr: "127.0.0.1:9092",
+				},
+			})
+		}
+		require.NotPanics(t, fn)
+	})
+
+	t.Run("open client adds new broker entries", func(t *testing.T) {
+		c := &client{
+			brokers: make(map[int32]*Broker),
+		}
+		fn := func() {
+			c.updateBroker([]*Broker{
+				{
+					id:   0,
+					addr: "127.0.0.1:9092",
+				},
+			})
+		}
+		require.NotPanics(t, fn)
+		require.Len(t, c.brokers, 1)
+		assert.Equal(t, 0, int(c.brokers[0].ID()))
+		assert.Equal(t, "127.0.0.1:9092", c.brokers[0].Addr())
+	})
+
+	t.Run("open client adds, updates and removes broker entries", func(t *testing.T) {
+		c := &client{
+			brokers: map[int32]*Broker{
+				0: {
+					id:   0,
+					addr: "127.0.0.1:9092",
+				},
+				1: {
+					id:   1,
+					addr: "127.0.0.1:9093",
+				},
+			},
+		}
+		fn := func() {
+			c.updateBroker([]*Broker{
+				{
+					id:   1,
+					addr: "127.0.0.1:19093", // new addr for existing broker
+				},
+				{
+					id:   2,
+					addr: "127.0.0.1:19094",
+				},
+			})
+		}
+		require.NotPanics(t, fn)
+		require.Len(t, c.brokers, 2)
+		assert.Equal(t, 1, int(c.brokers[1].ID()))
+		assert.Equal(t, "127.0.0.1:19093", c.brokers[1].Addr())
+		assert.Equal(t, 2, int(c.brokers[2].ID()))
+		assert.Equal(t, "127.0.0.1:19094", c.brokers[2].Addr())
+	})
+}