fix(proto): use up to V3 of OffsetRequest

We'd inadvertently listed up to V5 in the isValidVersion range, but
the V4 protocol changes were not implemented, so roll that back to V3,
but start using V2 and V3 where appropriate

Signed-off-by: Dominic Evans <[email protected]>

client.go

@@ -163,7 +163,6 @@ type client struct {
 	cachedPartitionsResults map[string][maxPartitionIndex][]int32
 
 	lock sync.RWMutex // protects access to the maps that hold cluster state.
-
 }
 
 // NewClient creates a new Client. It connects to one of the given broker addresses
@@ -899,9 +898,20 @@ func (client *client) getOffset(topic string, partitionID int32, time int64) (in
 	}
 
 	request := &OffsetRequest{}
+	// Version 1 removes MaxNumOffsets.  From this version forward, only a single
+	// offset can be returned.
 	if client.conf.Version.IsAtLeast(V0_10_1_0) {
 		request.Version = 1
 	}
+	// Version 2 adds the isolation level, which is used for transactional reads.
+	if client.conf.Version.IsAtLeast(V0_11_0_0) {
+		request.Version = 2
+	}
+	// Version 3 is the same as version 2.
+	if client.conf.Version.IsAtLeast(V2_0_0_0) {
+		request.Version = 3
+	}
+
 	request.AddBlock(topic, partitionID, time, 1)
 
 	response, err := broker.GetAvailableOffsets(request)

offset_request.go

@@ -138,23 +138,21 @@ func (r *OffsetRequest) headerVersion() int16 {
 }
 
 func (r *OffsetRequest) isValidVersion() bool {
-	return r.Version >= 0 && r.Version <= 5
+	return r.Version >= 0 && r.Version <= 3
 }
 
 func (r *OffsetRequest) requiredVersion() KafkaVersion {
 	switch r.Version {
-	case 5:
-		return V2_2_0_0
-	case 4:
-		return V2_1_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_8_2_0
 	default:
-		return MinVersion
+		return V2_0_0_0
 	}
 }
 

offset_response.go

@@ -168,23 +168,21 @@ func (r *OffsetResponse) headerVersion() int16 {
 }
 
 func (r *OffsetResponse) isValidVersion() bool {
-	return r.Version >= 0 && r.Version <= 5
+	return r.Version >= 0 && r.Version <= 3
 }
 
 func (r *OffsetResponse) requiredVersion() KafkaVersion {
 	switch r.Version {
-	case 5:
-		return V2_2_0_0
-	case 4:
-		return V2_1_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_8_2_0
 	default:
-		return MinVersion
+		return V2_0_0_0
 	}
 }