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())