chore(proto): permit DeleteGroupsRequest V1

admin.go

@@ -468,6 +468,9 @@ func (ca *clusterAdmin) CreatePartitions(topic string, count int32, assignment [
 		Timeout:         ca.conf.Admin.Timeout,
 		ValidateOnly:    validateOnly,
 	}
+	if ca.conf.Version.IsAtLeast(V2_0_0_0) {
+		request.Version = 1
+	}
 
 	return ca.retryOnError(isErrNoController, func() error {
 		b, err := ca.Controller()
@@ -713,6 +716,9 @@ func (ca *clusterAdmin) AlterConfig(resourceType ConfigResourceType, name string
 		Resources:    resources,
 		ValidateOnly: validateOnly,
 	}
+	if ca.conf.Version.IsAtLeast(V2_0_0_0) {
+		request.Version = 1
+	}
 
 	var (
 		b   *Broker
@@ -1027,6 +1033,9 @@ func (ca *clusterAdmin) DeleteConsumerGroup(group string) error {
 	request := &DeleteGroupsRequest{
 		Groups: []string{group},
 	}
+	if ca.conf.Version.IsAtLeast(V2_0_0_0) {
+		request.Version = 1
+	}
 
 	resp, err := coordinator.DeleteGroups(request)
 	if err != nil {

alter_configs_request.go

@@ -123,9 +123,16 @@ func (a *AlterConfigsRequest) headerVersion() int16 {
 }
 
 func (a *AlterConfigsRequest) isValidVersion() bool {
-	return a.Version == 0
+	return a.Version >= 0 && a.Version <= 1
 }
 
 func (a *AlterConfigsRequest) requiredVersion() KafkaVersion {
-	return V0_11_0_0
+	switch a.Version {
+	case 1:
+		return V2_0_0_0
+	case 0:
+		return V0_11_0_0
+	default:
+		return V2_0_0_0
+	}
 }

alter_configs_response.go

@@ -113,11 +113,18 @@ func (a *AlterConfigsResponse) headerVersion() int16 {
 }
 
 func (a *AlterConfigsResponse) isValidVersion() bool {
-	return a.Version == 0
+	return a.Version >= 0 && a.Version <= 1
 }
 
 func (a *AlterConfigsResponse) requiredVersion() KafkaVersion {
-	return V0_11_0_0
+	switch a.Version {
+	case 1:
+		return V2_0_0_0
+	case 0:
+		return V0_11_0_0
+	default:
+		return V2_0_0_0
+	}
 }
 
 func (r *AlterConfigsResponse) throttleTime() time.Duration {

create_partitions_request.go

@@ -73,11 +73,18 @@ func (r *CreatePartitionsRequest) headerVersion() int16 {
 }
 
 func (r *CreatePartitionsRequest) isValidVersion() bool {
-	return r.Version == 0
+	return r.Version >= 0 && r.Version <= 1
 }
 
 func (r *CreatePartitionsRequest) requiredVersion() KafkaVersion {
-	return V1_0_0_0
+	switch r.Version {
+	case 1:
+		return V2_0_0_0
+	case 0:
+		return V1_0_0_0
+	default:
+		return V2_0_0_0
+	}
 }
 
 type TopicPartition struct {

create_partitions_response.go

@@ -69,11 +69,18 @@ func (r *CreatePartitionsResponse) headerVersion() int16 {
 }
 
 func (r *CreatePartitionsResponse) isValidVersion() bool {
-	return r.Version == 0
+	return r.Version >= 0 && r.Version <= 1
 }
 
 func (r *CreatePartitionsResponse) requiredVersion() KafkaVersion {
-	return V1_0_0_0
+	switch r.Version {
+	case 1:
+		return V2_0_0_0
+	case 0:
+		return V1_0_0_0
+	default:
+		return V2_0_0_0
+	}
 }
 
 func (r *CreatePartitionsResponse) throttleTime() time.Duration {

delete_groups_request.go

@@ -27,11 +27,18 @@ func (r *DeleteGroupsRequest) headerVersion() int16 {
 }
 
 func (r *DeleteGroupsRequest) isValidVersion() bool {
-	return r.Version == 0
+	return r.Version >= 0 && r.Version <= 1
 }
 
 func (r *DeleteGroupsRequest) requiredVersion() KafkaVersion {
-	return V1_1_0_0
+	switch r.Version {
+	case 1:
+		return V2_0_0_0
+	case 0:
+		return V1_1_0_0
+	default:
+		return V2_0_0_0
+	}
 }
 
 func (r *DeleteGroupsRequest) AddGroup(group string) {

delete_groups_response.go

@@ -71,11 +71,18 @@ func (r *DeleteGroupsResponse) headerVersion() int16 {
 }
 
 func (r *DeleteGroupsResponse) isValidVersion() bool {
-	return r.Version == 0
+	return r.Version >= 0 && r.Version <= 1
 }
 
 func (r *DeleteGroupsResponse) requiredVersion() KafkaVersion {
-	return V1_1_0_0
+	switch r.Version {
+	case 1:
+		return V2_0_0_0
+	case 0:
+		return V1_1_0_0
+	default:
+		return V2_0_0_0
+	}
 }
 
 func (r *DeleteGroupsResponse) throttleTime() time.Duration {

describe_configs_request.go

@@ -109,11 +109,13 @@ func (r *DescribeConfigsRequest) isValidVersion() bool {
 
 func (r *DescribeConfigsRequest) requiredVersion() KafkaVersion {
 	switch r.Version {
-	case 1:
-		return V1_1_0_0
 	case 2:
 		return V2_0_0_0
-	default:
+	case 1:
+		return V1_1_0_0
+	case 0:
 		return V0_11_0_0
+	default:
+		return V2_0_0_0
 	}
 }

describe_configs_response.go

@@ -122,12 +122,14 @@ func (r *DescribeConfigsResponse) isValidVersion() bool {
 
 func (r *DescribeConfigsResponse) requiredVersion() KafkaVersion {
 	switch r.Version {
-	case 1:
-		return V1_0_0_0
 	case 2:
 		return V2_0_0_0
-	default:
+	case 1:
+		return V1_1_0_0
+	case 0:
 		return V0_11_0_0
+	default:
+		return V2_0_0_0
 	}
 }
 

request_test.go

@@ -241,6 +241,7 @@ func TestAllocateBodyProtocolVersions(t *testing.T) {
 				resp := allocateResponseBody(req)
 				assert.NotNil(t, resp, fmt.Sprintf("%s has no matching response type in allocateResponseBody", reflect.TypeOf(req)))
 				assert.Equal(t, req.isValidVersion(), resp.isValidVersion(), fmt.Sprintf("%s isValidVersion should match %s", reflect.TypeOf(req), reflect.TypeOf(resp)))
+				assert.Equal(t, req.requiredVersion(), resp.requiredVersion(), fmt.Sprintf("%s requiredVersion should match %s", reflect.TypeOf(req), reflect.TypeOf(resp)))
 				for _, body := range []protocolBody{req, resp} {
 					assert.Equal(t, key, body.key())
 					assert.Equal(t, version, body.version())