InvalidRequest after TopicAlreadyExists was not thrown while creating topics.

[Y-Wakuta]

Describe the bug

When creating topics, if TopicConfigs with specific order are passed to Conn.CreateTopics(), InvalidRequests in createTopicsResponseV0 are ignored.This made it difficult to detect some topics are not created due to InvalidRequest. In detail, this problem occurs when the passed TopicConfigs are like [topic A (already existing), topic B, topic B]. In this case, since there are two entries for topic B, InvalidRequest is expected to returned for the second and third TopicConfig. However, we receive TopicAlreadyExists error for topic A and exits Conn.createTopics(). And then Conn.CreateTopics() ignores TopicAlreadyExits error and returns nil (InvalidRequest is ignored, and not thrown).

Kafka Version

• kafka version: 2.3.1
• kafka-go version: v0.4.39

To Reproduce

1. setup kafka environment

$ git clone https://github.com/segmentio/kafka-go.git
$ cd kafka-go
$ docker-compose up

2. create topics

package main

import (
	"context"
	"fmt"
	"github.com/segmentio/kafka-go"
)

func main() {
	conn, err := kafka.DialContext(context.Background(), "tcp", "localhost:9092")
	defer conn.Close()
	if err != nil {
		panic(err)
	}

	preCreatedTopicName := "topic_A"
	err = conn.CreateTopics(createTopicConfig(preCreatedTopicName))
	if err != nil {
		panic(err)
	}

	newlyCreatedTopicName := "topic_B"
	err = conn.CreateTopics( // Three TopicConfigs below contains duplicate topic names and expected to throw InvalidRequest
		createTopicConfig(preCreatedTopicName), // Creating this topic conceals InvalidRequest.
		createTopicConfig(newlyCreatedTopicName),
		createTopicConfig(newlyCreatedTopicName),
	)
	if err != nil {
		panic(err)
	}
	fmt.Println("finished")
}

func createTopicConfig(topicName string) kafka.TopicConfig {
	return kafka.TopicConfig{
		Topic:             topicName,
		NumPartitions:     1,
		ReplicationFactor: 1,
	}
}

Expected Behavior

InvalidRequest is returned from conn.CreateTopics() when the given TopicsConfigs contain duplicated TopicConfig (e.g. topic_B in the above example).

Observed Behavior

Conn.CreateTopics() does not return an InvalidRequest and then conn.CreateTopic() successees.

Additional Context

The cause of this behavior is Conn.createTopics() exits for the first non-nil error (in the above example, TopicAlreadyExists), at here. And then, Conn.CreateTopics() ignores TopicAlreadyExists and returns nil at here. I suggest move TopicAlreadyExists check from CreateTopics() to createTopics() like this (If this change makes sence, I am willing to submit PR).