fix(proto): use full range of MetadataRequest

Even though most of these are identical, we may as well match up
correctly.

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

Author: dnwe

metadata_request.go

@@ -17,6 +17,10 @@ func NewMetadataRequest(version KafkaVersion, topics []string) *MetadataRequest
 		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
 	}
@@ -94,19 +98,21 @@ func (r *MetadataRequest) isValidVersion() bool {
 
 func (r *MetadataRequest) requiredVersion() KafkaVersion {
 	switch r.Version {
-	case 1:
-		return V0_10_0_0
-	case 2:
-		return V0_10_1_0
-	case 3, 4:
-		return V0_11_0_0
-	case 5:
-		return V1_0_0_0
-	case 6:
-		return V2_0_0_0
 	case 7:
 		return V2_1_0_0
+	case 6:
+		return V2_0_0_0
+	case 5:
+		return V1_0_0_0
+	case 3, 4:
+		return V0_11_0_0
+	case 2:
+		return V0_10_1_0
+	case 1:
+		return V0_10_0_0
+	case 0:
+		return V0_8_2_0
 	default:
-		return MinVersion
+		return V2_1_0_0
 	}
 }

metadata_response.go

@@ -281,20 +281,22 @@ func (r *MetadataResponse) isValidVersion() bool {
 
 func (r *MetadataResponse) requiredVersion() KafkaVersion {
 	switch r.Version {
-	case 1:
-		return V0_10_0_0
-	case 2:
-		return V0_10_1_0
-	case 3, 4:
-		return V0_11_0_0
-	case 5:
-		return V1_0_0_0
-	case 6:
-		return V2_0_0_0
 	case 7:
 		return V2_1_0_0
+	case 6:
+		return V2_0_0_0
+	case 5:
+		return V1_0_0_0
+	case 3, 4:
+		return V0_11_0_0
+	case 2:
+		return V0_10_1_0
+	case 1:
+		return V0_10_0_0
+	case 0:
+		return V0_8_2_0
 	default:
-		return MinVersion
+		return V2_1_0_0
 	}
 }