feat(protocol): support CreateTopicRequest V4 (#3238)

dnwe · web-flow · commit 1c971aec3deb · 2025-08-06T13:12:36.000+01:00
Add support for using CreateTopicRequest V4 against Kafka 2.4 and newer clusters. Protocol-wise this is identical to the existing V3, but the version indicates that the server should accept `-1` as the partition and replicate-factor values to indicate "use the server side defaults" (KIP-464). Contributes-to: #3233 Signed-off-by: Dominic Evans <dominic.evans@uk.ibm.com>
diff --git a/admin.go b/admin.go
@@ -251,22 +251,12 @@ func (ca *clusterAdmin) CreateTopic(topic string, detail *TopicDetail, validateO
 	topicDetails := make(map[string]*TopicDetail)
 	topicDetails[topic] = detail
 
-	request := &CreateTopicsRequest{
-		TopicDetails: topicDetails,
-		ValidateOnly: validateOnly,
-		Timeout:      ca.conf.Admin.Timeout,
-	}
-
-	if ca.conf.Version.IsAtLeast(V2_0_0_0) {
-		// Version 3 is the same as version 2 (brokers response before throttling)
-		request.Version = 3
-	} else if ca.conf.Version.IsAtLeast(V0_11_0_0) {
-		// Version 2 is the same as version 1 (response has ThrottleTime)
-		request.Version = 2
-	} else if ca.conf.Version.IsAtLeast(V0_10_2_0) {
-		// Version 1 adds validateOnly.
-		request.Version = 1
-	}
+	request := NewCreateTopicsRequest(
+		ca.conf.Version,
+		topicDetails,
+		ca.conf.Admin.Timeout,
+		validateOnly,
+	)
 
 	return ca.retryOnError(isRetriableControllerError, func() error {
 		b, err := ca.Controller()
diff --git a/create_topics_request.go b/create_topics_request.go
@@ -20,16 +20,29 @@ func (c *CreateTopicsRequest) setVersion(v int16) {
 	c.Version = v
 }
 
-func NewCreateTopicsRequest(version KafkaVersion, topicDetails map[string]*TopicDetail, timeout time.Duration) *CreateTopicsRequest {
+func NewCreateTopicsRequest(
+	version KafkaVersion,
+	topicDetails map[string]*TopicDetail,
+	timeout time.Duration,
+	validateOnly bool,
+) *CreateTopicsRequest {
 	r := &CreateTopicsRequest{
 		TopicDetails: topicDetails,
 		Timeout:      timeout,
-	}
-	if version.IsAtLeast(V2_0_0_0) {
+		ValidateOnly: validateOnly,
+	}
+	switch {
+	case version.IsAtLeast(V2_4_0_0):
+		// Version 4 makes partitions/replicationFactor optional even when assignments are not present (KIP-464)
+		r.Version = 4
+	case version.IsAtLeast(V2_0_0_0):
+		// Version 3 is the same as version 2 (brokers response before throttling)
 		r.Version = 3
-	} else if version.IsAtLeast(V0_11_0_0) {
+	case version.IsAtLeast(V0_11_0_0):
+		// Version 2 is the same as version 1 (response has ThrottleTime)
 		r.Version = 2
-	} else if version.IsAtLeast(V0_10_2_0) {
+	case version.IsAtLeast(V0_10_2_0):
+		// Version 1 adds validateOnly.
 		r.Version = 1
 	}
 	return r
@@ -102,16 +115,18 @@ func (c *CreateTopicsRequest) version() int16 {
 	return c.Version
 }
 
-func (r *CreateTopicsRequest) headerVersion() int16 {
+func (c *CreateTopicsRequest) headerVersion() int16 {
 	return 1
 }
 
 func (c *CreateTopicsRequest) isValidVersion() bool {
-	return c.Version >= 0 && c.Version <= 3
+	return c.Version >= 0 && c.Version <= 4
 }
 
 func (c *CreateTopicsRequest) requiredVersion() KafkaVersion {
 	switch c.Version {
+	case 4:
+		return V2_4_0_0
 	case 3:
 		return V2_0_0_0
 	case 2:
diff --git a/functional_test.go b/functional_test.go
@@ -387,7 +387,7 @@ func prepareTestTopics(ctx context.Context, env *testEnvironment) error {
 
 	// now create the topics empty
 	{
-		request := NewCreateTopicsRequest(config.Version, testTopicDetails, time.Minute)
+		request := NewCreateTopicsRequest(config.Version, testTopicDetails, time.Minute, false)
 		createRes, err := controller.CreateTopics(request)
 		if err != nil {
 			return fmt.Errorf("failed to create test topics: %w", err)

Original file line number	Diff line number	Diff line change
`@@ -387,7 +387,7 @@ func prepareTestTopics(ctx context.Context, env *testEnvironment) error {`
`387`	`387`
`388`	`388`	`// now create the topics empty`
`389`	`389`	`{`
`390`		`- request := NewCreateTopicsRequest(config.Version, testTopicDetails, time.Minute)`
	`390`	`+ request := NewCreateTopicsRequest(config.Version, testTopicDetails, time.Minute, false)`
`391`	`391`	`createRes, err := controller.CreateTopics(request)`
`392`	`392`	`if err != nil {`
`393`	`393`	`return fmt.Errorf("failed to create test topics: %w", err)`