cherry-pick some PRs 2 round

admin.go

@@ -700,11 +700,8 @@ func (ca *clusterAdmin) DescribeConfig(resource ConfigResource) ([]ConfigEntry,
 
 	for _, rspResource := range rsp.Resources {
 		if rspResource.Name == resource.Name {
-			if rspResource.ErrorMsg != "" {
-				return nil, errors.New(rspResource.ErrorMsg)
-			}
 			if rspResource.ErrorCode != 0 {
-				return nil, KError(rspResource.ErrorCode)
+				return nil, &DescribeConfigError{Err: KError(rspResource.ErrorCode), ErrMsg: rspResource.ErrorMsg}
 			}
 			for _, cfgEntry := range rspResource.Configs {
 				entries = append(entries, *cfgEntry)
@@ -758,11 +755,8 @@ func (ca *clusterAdmin) AlterConfig(resourceType ConfigResourceType, name string
 
 	for _, rspResource := range rsp.Resources {
 		if rspResource.Name == name {
-			if rspResource.ErrorMsg != "" {
-				return errors.New(rspResource.ErrorMsg)
-			}
 			if rspResource.ErrorCode != 0 {
-				return KError(rspResource.ErrorCode)
+				return &AlterConfigError{Err: KError(rspResource.ErrorCode), ErrMsg: rspResource.ErrorMsg}
 			}
 		}
 	}

alter_configs_response.go

@@ -1,6 +1,9 @@
 package sarama
 
-import "time"
+import (
+	"fmt"
+	"time"
+)
 
 // AlterConfigsResponse is a response type for alter config
 type AlterConfigsResponse struct {
@@ -9,6 +12,19 @@ type AlterConfigsResponse struct {
 	Resources    []*AlterConfigsResourceResponse
 }
 
+type AlterConfigError struct {
+	Err    KError
+	ErrMsg string
+}
+
+func (c *AlterConfigError) Error() string {
+	text := c.Err.Error()
+	if c.ErrMsg != "" {
+		text = fmt.Sprintf("%s - %s", text, c.ErrMsg)
+	}
+	return text
+}
+
 // AlterConfigsResourceResponse is a response type for alter config resource
 type AlterConfigsResourceResponse struct {
 	ErrorCode int16

async_producer.go

@@ -1350,7 +1350,7 @@ func (bp *brokerProducer) handleSuccess(sent *produceSet, response *ProduceRespo
 			bp.parent.returnSuccesses(pSet.msgs)
 		// Retriable errors
 		case ErrInvalidMessage, ErrUnknownTopicOrPartition, ErrLeaderNotAvailable, ErrNotLeaderForPartition,
-			ErrRequestTimedOut, ErrNotEnoughReplicas, ErrNotEnoughReplicasAfterAppend:
+			ErrRequestTimedOut, ErrNotEnoughReplicas, ErrNotEnoughReplicasAfterAppend, ErrKafkaStorageError:
 			if bp.parent.conf.Producer.Retry.Max <= 0 {
 				bp.parent.abandonBrokerConnection(bp.broker)
 				bp.parent.returnErrors(pSet.msgs, block.Err)
@@ -1389,7 +1389,7 @@ func (bp *brokerProducer) handleSuccess(sent *produceSet, response *ProduceRespo
 
 			switch block.Err {
 			case ErrInvalidMessage, ErrUnknownTopicOrPartition, ErrLeaderNotAvailable, ErrNotLeaderForPartition,
-				ErrRequestTimedOut, ErrNotEnoughReplicas, ErrNotEnoughReplicasAfterAppend:
+				ErrRequestTimedOut, ErrNotEnoughReplicas, ErrNotEnoughReplicasAfterAppend, ErrKafkaStorageError:
 				Logger.Printf("producer/broker/%d state change to [retrying] on %s/%d because %v\n",
 					bp.broker.ID(), topic, partition, block.Err)
 				if bp.currentRetries[topic] == nil {

client.go

@@ -721,9 +721,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

@@ -10,6 +10,8 @@ import (
 	"time"
 
 	"github.com/rcrowley/go-metrics"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func safeClose(t testing.TB, c io.Closer) {
@@ -1249,3 +1251,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())
+	})
+}

describe_configs_response.go

@@ -34,6 +34,19 @@ const (
 	SourceDefault
 )
 
+type DescribeConfigError struct {
+	Err    KError
+	ErrMsg string
+}
+
+func (c *DescribeConfigError) Error() string {
+	text := c.Err.Error()
+	if c.ErrMsg != "" {
+		text = fmt.Sprintf("%s - %s", text, c.ErrMsg)
+	}
+	return text
+}
+
 type DescribeConfigsResponse struct {
 	Version      int16
 	ThrottleTime time.Duration