make InitProducerID return ErrOutOfBrokers with the the per-broker er…

…rors wrapped

Signed-off-by: kwall <[email protected]>

client.go

@@ -212,23 +212,24 @@ func (client *client) Broker(brokerID int32) (*Broker, error) {
 }
 
 func (client *client) InitProducerID() (*InitProducerIDResponse, error) {
-	err := ErrOutOfBrokers
+	brokerErrors := make([]error, 0)
 	for broker := client.any(); broker != nil; broker = client.any() {
 		var response *InitProducerIDResponse
 		req := &InitProducerIDRequest{}
 
-		response, err = broker.InitProducerID(req)
+		response, err := broker.InitProducerID(req)
 		if err == nil {
 			return response, nil
 		} else {
 			// some error, remove that broker and try again
 			Logger.Printf("Client got error from broker %d when issuing InitProducerID : %v\n", broker.ID(), err)
 			_ = broker.Close()
+			brokerErrors = append(brokerErrors, err)
 			client.deregisterBroker(broker)
 		}
 	}
 
-	return nil, err
+	return nil, Wrap(ErrOutOfBrokers, brokerErrors...)
 }
 
 func (client *client) Close() error {

client_test.go

@@ -1038,3 +1038,34 @@ func TestClientCoordinatorConnectionRefused(t *testing.T) {
 
 	safeClose(t, client)
 }
+
+func TestInitProducerIDConnectionRefused(t *testing.T) {
+	t.Parallel()
+	seedBroker := NewMockBroker(t, 1)
+	seedBroker.Returns(&MetadataResponse{Version: 1})
+
+	config := NewTestConfig()
+	config.Producer.Idempotent = true
+	config.Version = V0_11_0_0
+	config.Producer.RequiredAcks = WaitForAll
+	config.Net.MaxOpenRequests = 1
+
+	client, err := NewClient([]string{seedBroker.Addr()}, config)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	seedBroker.Close()
+
+	_, err = client.InitProducerID()
+
+	if !errors.Is(err, ErrOutOfBrokers) {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	if !errors.Is(err, io.EOF) {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	safeClose(t, client)
+}