AdminClient: Default parameters support for AdminClient#createTopic (#616)

This is needed because librdkafka supports
AdminClient.CreateTopics(...) with default value(-1) of NumPartitions
and ReplicationFactor, but currently it's not possible to pass a value
of -1 or 0 for ReplicationFactor to CreateTopics without specifying an
explicit ReplicaAssignment.

Author: jliunyu

CHANGELOG.md

@@ -1,5 +1,15 @@
 # Confluent's Golang client for Apache Kafka
 
+## v1.7.0
+
+### Fixes
+
+* AdminClient.CreateTopics() previously did not accept default value(-1) of
+  ReplicationFactor without specifying an explicit ReplicaAssignment, this is
+  now fixed.
+
+
+
 ## v1.6.1
 
 v1.6.1 is a feature release:

kafka/adminapi.go

@@ -514,10 +514,6 @@ func (a *AdminClient) CreateTopics(ctx context.Context, topics []TopicSpecificat
 				return nil, newErrorFromString(ErrInvalidArg,
 					"TopicSpecification.ReplicaAssignment must contain exactly TopicSpecification.NumPartitions partitions")
 			}
-
-		} else if cReplicationFactor == -1 {
-			return nil, newErrorFromString(ErrInvalidArg,
-				"TopicSpecification.ReplicationFactor or TopicSpecification.ReplicaAssignment must be specified")
 		}
 
 		cTopics[i] = C.rd_kafka_NewTopic_new(

kafka/adminapi_test.go

@@ -23,6 +23,59 @@ import (
 	"time"
 )
 
+// TestAdminAPIWithDefaultValue tests CreateTopics with default
+// NumPartitions and ReplicationFactor values
+func TestAdminAPIWithDefaultValue(t *testing.T) {
+	if !testconfRead() {
+		t.Skipf("Missing testconf.json")
+	}
+
+	topic := "testWithDefaultValue"
+
+	conf := ConfigMap{"bootstrap.servers": testconf.Brokers}
+	if err := conf.updateFromTestconf(); err != nil {
+		t.Fatalf("Failed to update test configuration: %v\n", err)
+	}
+
+	expDuration, err := time.ParseDuration("30s")
+	if err != nil {
+		t.Fatalf("Failed to Parse Duration: %s", err)
+	}
+
+	adminClient, err := NewAdminClient(&conf)
+	if err != nil {
+		t.Fatalf("Failed to create AdminClient %v", err)
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), expDuration)
+	defer cancel()
+	res, err := adminClient.CreateTopics(
+		ctx,
+		[]TopicSpecification{
+			{
+				Topic:             topic,
+				NumPartitions:     -1,
+				ReplicationFactor: -1,
+			},
+		})
+	if err != nil {
+		adminClient.Close()
+		t.Fatalf("Failed to create topics %v\n", err)
+	}
+	t.Logf("Succeed to create topic %v\n", res)
+
+	ctx, cancel = context.WithTimeout(context.Background(), expDuration)
+	defer cancel()
+	res, err = adminClient.DeleteTopics(ctx, []string{topic})
+	if err != nil {
+		adminClient.Close()
+		t.Fatalf("Failed to delete topic %v, err: %v", topic, err)
+	}
+	t.Logf("Succeed to delete topic %v\n", res)
+
+	adminClient.Close()
+}
+
 func testAdminAPIs(what string, a *AdminClient, t *testing.T) {
 	t.Logf("AdminClient API testing on %s: %s", a, what)