fix: update constructors to use getMaxAPIVersionForKafkaVersion

Signed-off-by: Giorgio Pellero <[email protected]>

Author: trapped

create_topics_request.go

@@ -20,13 +20,7 @@ func NewCreateTopicsRequest(version KafkaVersion, topicDetails map[string]*Topic
 	r := &CreateTopicsRequest{
 		TopicDetails: topicDetails,
 		Timeout:      timeout,
-	}
-	if version.IsAtLeast(V2_0_0_0) {
-		r.Version = 3
-	} else if version.IsAtLeast(V0_11_0_0) {
-		r.Version = 2
-	} else if version.IsAtLeast(V0_10_2_0) {
-		r.Version = 1
+		Version:      getMaxAPIVersionForKafkaVersion(APIKeyCreateTopics, version),
 	}
 	return r
 }

delete_topics_request.go

@@ -12,13 +12,7 @@ func NewDeleteTopicsRequest(version KafkaVersion, topics []string, timeout time.
 	d := &DeleteTopicsRequest{
 		Topics:  topics,
 		Timeout: timeout,
-	}
-	if version.IsAtLeast(V2_1_0_0) {
-		d.Version = 3
-	} else if version.IsAtLeast(V2_0_0_0) {
-		d.Version = 2
-	} else if version.IsAtLeast(V0_11_0_0) {
-		d.Version = 1
+		Version: getMaxAPIVersionForKafkaVersion(APIKeyDeleteTopics, version),
 	}
 	return d
 }

metadata_request.go

@@ -22,25 +22,9 @@ type MetadataRequest struct {
 }
 
 func NewMetadataRequest(version KafkaVersion, topics []string) *MetadataRequest {
-	m := &MetadataRequest{Topics: topics}
-	if version.IsAtLeast(V2_8_0_0) {
-		m.Version = 10
-	} else if version.IsAtLeast(V2_4_0_0) {
-		m.Version = 9
-	} else if version.IsAtLeast(V2_4_0_0) {
-		m.Version = 8
-	} else if version.IsAtLeast(V2_1_0_0) {
-		m.Version = 7
-	} else if version.IsAtLeast(V2_0_0_0) {
-		m.Version = 6
-	} else if version.IsAtLeast(V1_0_0_0) {
-		m.Version = 5
-	} else if version.IsAtLeast(V0_11_0_0) {
-		m.Version = 4
-	} else if version.IsAtLeast(V0_10_1_0) {
-		m.Version = 2
-	} else if version.IsAtLeast(V0_10_0_0) {
-		m.Version = 1
+	m := &MetadataRequest{
+		Topics:  topics,
+		Version: getMaxAPIVersionForKafkaVersion(APIKeyMetadata, version),
 	}
 	return m
 }

offset_fetch_request.go

@@ -12,33 +12,18 @@ func NewOffsetFetchRequest(
 	group string,
 	partitions map[string][]int32,
 ) *OffsetFetchRequest {
+	// Version history:
+	// - Version 7: Adding the require stable flag (V2_5_0_0+)
+	// - Version 6: First flexible version (V2_4_0_0+)
+	// - Versions 3, 4, and 5: Same as version 2 (V2_1_0_0+)
+	// - Version 2: Request can contain null topics array to indicate offsets for all topics
+	//   should be fetched. Also returns top level error code for group or coordinator level errors (V0_10_2_0+)
+	// - Version 1: Broker supports fetching offsets from internal __consumer_offsets topic (V0_8_2_0+)
+	// - Version 0: Request read offsets from ZK
 	request := &OffsetFetchRequest{
 		ConsumerGroup: group,
 		partitions:    partitions,
-	}
-	if version.IsAtLeast(V2_5_0_0) {
-		// Version 7 is adding the require stable flag.
-		request.Version = 7
-	} else if version.IsAtLeast(V2_4_0_0) {
-		// Version 6 is the first flexible version.
-		request.Version = 6
-	} else if version.IsAtLeast(V2_1_0_0) {
-		// Version 3, 4, and 5 are the same as version 2.
-		request.Version = 5
-	} else if version.IsAtLeast(V2_0_0_0) {
-		request.Version = 4
-	} else if version.IsAtLeast(V0_11_0_0) {
-		request.Version = 3
-	} else if version.IsAtLeast(V0_10_2_0) {
-		// Starting in version 2, the request can contain a null topics array to indicate that offsets
-		// for all topics should be fetched. It also returns a top level error code
-		// for group or coordinator level errors.
-		request.Version = 2
-	} else if version.IsAtLeast(V0_8_2_0) {
-		// In version 0, the request read offsets from ZK.
-		//
-		// Starting in version 1, the broker supports fetching offsets from the internal __consumer_offsets topic.
-		request.Version = 1
+		Version:       getMaxAPIVersionForKafkaVersion(APIKeyOffsetFetch, version),
 	}
 
 	return request