fix(proto): correct JoinGroup usage for wider version range

Signed-off-by: Dominic Evans <dominic.evans@uk.ibm.com>

consumer_group.go

@@ -433,7 +433,24 @@ func (c *consumerGroup) joinGroupRequest(coordinator *Broker, topics []string) (
 		req.Version = 1
 		req.RebalanceTimeout = int32(c.config.Consumer.Group.Rebalance.Timeout / time.Millisecond)
 	}
-	if c.groupInstanceId != nil {
+	if c.config.Version.IsAtLeast(V0_11_0_0) {
+		req.Version = 2
+	}
+	if c.config.Version.IsAtLeast(V0_11_0_0) {
+		req.Version = 2
+	}
+	if c.config.Version.IsAtLeast(V2_0_0_0) {
+		req.Version = 3
+	}
+	// XXX: protocol states "Starting from version 4, the client needs to issue a
+	// second request to join group", so not enabling this until we can
+	// investigate
+	/*
+		if c.config.Version.IsAtLeast(V2_2_0_0) {
+			req.Version = 4
+		}
+	*/
+	if c.config.Version.IsAtLeast(V2_3_0_0) {
 		req.Version = 5
 		req.GroupInstanceId = c.groupInstanceId
 	}

join_group_request.go

@@ -156,14 +156,20 @@ func (r *JoinGroupRequest) isValidVersion() bool {
 
 func (r *JoinGroupRequest) requiredVersion() KafkaVersion {
 	switch r.Version {
-	case 4, 5:
+	case 5:
 		return V2_3_0_0
-	case 2, 3:
+	case 4:
+		return V2_2_0_0
+	case 3:
+		return V2_0_0_0
+	case 2:
 		return V0_11_0_0
 	case 1:
 		return V0_10_1_0
+	case 0:
+		return V0_10_0_0
 	default:
-		return V0_9_0_0
+		return V2_3_0_0
 	}
 }
 

join_group_response.go

@@ -153,14 +153,20 @@ func (r *JoinGroupResponse) isValidVersion() bool {
 
 func (r *JoinGroupResponse) requiredVersion() KafkaVersion {
 	switch r.Version {
-	case 3, 4, 5:
+	case 5:
 		return V2_3_0_0
+	case 4:
+		return V2_2_0_0
+	case 3:
+		return V2_0_0_0
 	case 2:
 		return V0_11_0_0
 	case 1:
 		return V0_10_1_0
+	case 0:
+		return V0_10_0_0
 	default:
-		return V0_9_0_0
+		return V2_3_0_0
 	}
 }