Repository: segmentio/kafka-go (language: go)

Issue:
InvalidRequest after TopicAlreadyExists was not thrown while creating topics.
**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

```shell
$ git clone https://github.com/segmentio/kafka-go.git
$ cd kafka-go
$ docker-compose up
```

2. create topics

```go
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](https://github.com/segmentio/kafka-go/blob/f6986fb02ee7f8c6d6708da3f44b7ccf28fa435b/createtopics.go#L367). And then, `Conn.CreateTopics()` ignores TopicAlreadyExists and returns nil at [here](https://github.com/segmentio/kafka-go/blob/f6986fb02ee7f8c6d6708da3f44b7ccf28fa435b/createtopics.go#L391). I suggest move TopicAlreadyExists check from CreateTopics() to createTopics() like [this](https://github.com/Y-Wakuta/kafka-go/pull/1/files) (If this change makes sence, I am willing to submit PR).


Discussion / hints:

Fix patch:
diff --git a/createtopics.go b/createtopics.go
index 561425a53..8ad9ebf44 100644
--- a/createtopics.go
+++ b/createtopics.go
@@ -3,7 +3,6 @@ package kafka
 import (
 	"bufio"
 	"context"
-	"errors"
 	"fmt"
 	"net"
 	"time"
@@ -65,7 +64,6 @@ func (c *Client) CreateTopics(ctx context.Context, req *CreateTopicsRequest) (*C
 		TimeoutMs:    c.timeoutMs(ctx, defaultCreateTopicsTimeout),
 		ValidateOnly: req.ValidateOnly,
 	})
-
 	if err != nil {
 		return nil, fmt.Errorf("kafka.(*Client).CreateTopics: %w", err)
 	}
@@ -363,6 +361,9 @@ func (c *Conn) createTopics(request createTopicsRequestV0) (createTopicsResponse
 		return response, err
 	}
 	for _, tr := range response.TopicErrors {
+		if tr.ErrorCode == int16(TopicAlreadyExists) {
+			continue
+		}
 		if tr.ErrorCode != 0 {
 			return response, Error(tr.ErrorCode)
 		}
@@ -385,14 +386,5 @@ func (c *Conn) CreateTopics(topics ...TopicConfig) error {
 	_, err := c.createTopics(createTopicsRequestV0{
 		Topics: requestV0Topics,
 	})
-	if err != nil {
-		if errors.Is(err, TopicAlreadyExists) {
-			// ok
-			return nil
-		}
-
-		return err
-	}
-
-	return nil
+	return err
 }


Test patch:
diff --git a/createtopics_test.go b/createtopics_test.go
index 71cf456e0..38819c382 100644
--- a/createtopics_test.go
+++ b/createtopics_test.go
@@ -4,10 +4,86 @@ import (
 	"bufio"
 	"bytes"
 	"context"
+	"errors"
+	"net"
 	"reflect"
+	"strconv"
 	"testing"
 )
 
+func TestConnCreateTopics(t *testing.T) {
+	topic1 := makeTopic()
+	topic2 := makeTopic()
+
+	conn, err := DialContext(context.Background(), "tcp", "localhost:9092")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	defer func() {
+		err := conn.Close()
+		if err != nil {
+			t.Fatalf("failed to close connection: %v", err)
+		}
+	}()
+
+	controller, _ := conn.Controller()
+
+	controllerConn, err := Dial("tcp", net.JoinHostPort(controller.Host, strconv.Itoa(controller.Port)))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer controllerConn.Close()
+
+	err = controllerConn.CreateTopics(TopicConfig{
+		Topic:             topic1,
+		NumPartitions:     1,
+		ReplicationFactor: 1,
+	})
+	if err != nil {
+		t.Fatalf("unexpected error creating topic: %s", err.Error())
+	}
+
+	err = controllerConn.CreateTopics(TopicConfig{
+		Topic:             topic1,
+		NumPartitions:     1,
+		ReplicationFactor: 1,
+	})
+
+	// Duplicate topic should not return an error
+	if err != nil {
+		t.Fatalf("unexpected error creating duplicate topic topic: %v", err)
+	}
+
+	err = controllerConn.CreateTopics(
+		TopicConfig{
+			Topic:             topic1,
+			NumPartitions:     1,
+			ReplicationFactor: 1,
+		},
+		TopicConfig{
+			Topic:             topic2,
+			NumPartitions:     1,
+			ReplicationFactor: 1,
+		},
+		TopicConfig{
+			Topic:             topic2,
+			NumPartitions:     1,
+			ReplicationFactor: 1,
+		},
+	)
+
+	if err == nil {
+		t.Fatal("CreateTopics should have returned an error for invalid requests")
+	}
+
+	if !errors.Is(err, InvalidRequest) {
+		t.Fatalf("expected invalid request: %v", err)
+	}
+
+	deleteTopic(t, topic1)
+}
+
 func TestClientCreateTopics(t *testing.T) {
 	const (
 		topic1 = "client-topic-1"
@@ -59,7 +135,6 @@ func TestClientCreateTopics(t *testing.T) {
 			},
 		},
 	})
-
 	if err != nil {
 		t.Fatal(err)
 	}