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

async_producer_test.go

@@ -1111,7 +1111,7 @@ func TestAsyncProducerIdempotentGoldenPath(t *testing.T) {
 	broker := NewMockBroker(t, 1)
 
 	metadataResponse := &MetadataResponse{
-		Version:      1,
+		Version:      4,
 		ControllerID: 1,
 	}
 	metadataResponse.AddBroker(broker.Addr(), broker.BrokerID())
@@ -1169,7 +1169,7 @@ func TestAsyncProducerIdempotentRetryCheckBatch(t *testing.T) {
 		broker := NewMockBroker(t, 1)
 
 		metadataResponse := &MetadataResponse{
-			Version:      1,
+			Version:      4,
 			ControllerID: 1,
 		}
 		metadataResponse.AddBroker(broker.Addr(), broker.BrokerID())
@@ -1307,7 +1307,7 @@ func TestAsyncProducerIdempotentRetryCheckBatch_2378(t *testing.T) {
 	broker := NewMockBroker(t, 1)
 
 	metadataResponse := &MetadataResponse{
-		Version:      1,
+		Version:      4,
 		ControllerID: 1,
 	}
 	metadataResponse.AddBroker(broker.Addr(), broker.BrokerID())
@@ -1375,7 +1375,7 @@ func TestAsyncProducerIdempotentErrorOnOutOfSeq(t *testing.T) {
 	broker := NewMockBroker(t, 1)
 
 	metadataResponse := &MetadataResponse{
-		Version:      1,
+		Version:      4,
 		ControllerID: 1,
 	}
 	metadataResponse.AddBroker(broker.Addr(), broker.BrokerID())
@@ -1425,7 +1425,7 @@ func TestAsyncProducerIdempotentEpochRollover(t *testing.T) {
 	defer broker.Close()
 
 	metadataResponse := &MetadataResponse{
-		Version:      1,
+		Version:      4,
 		ControllerID: 1,
 	}
 	metadataResponse.AddBroker(broker.Addr(), broker.BrokerID())
@@ -1501,7 +1501,7 @@ func TestAsyncProducerIdempotentEpochExhaustion(t *testing.T) {
 	)
 
 	metadataResponse := &MetadataResponse{
-		Version:      1,
+		Version:      4,
 		ControllerID: 1,
 	}
 	metadataResponse.AddBroker(broker.Addr(), broker.BrokerID())
@@ -1737,7 +1737,7 @@ func TestTxmngInitProducerId(t *testing.T) {
 	defer broker.Close()
 
 	metadataLeader := new(MetadataResponse)
-	metadataLeader.Version = 1
+	metadataLeader.Version = 4
 	metadataLeader.AddBroker(broker.Addr(), broker.BrokerID())
 	broker.Returns(metadataLeader)
 
@@ -1879,7 +1879,7 @@ func TestTxnProduceRecordWithCommit(t *testing.T) {
 	config.Net.MaxOpenRequests = 1
 
 	metadataLeader := new(MetadataResponse)
-	metadataLeader.Version = 1
+	metadataLeader.Version = 4
 	metadataLeader.ControllerID = broker.brokerID
 	metadataLeader.AddBroker(broker.Addr(), broker.BrokerID())
 	metadataLeader.AddTopic("test-topic", ErrNoError)
@@ -1960,7 +1960,7 @@ func TestTxnProduceBatchAddPartition(t *testing.T) {
 	config.Producer.Partitioner = NewManualPartitioner
 
 	metadataLeader := new(MetadataResponse)
-	metadataLeader.Version = 1
+	metadataLeader.Version = 4
 	metadataLeader.ControllerID = broker.brokerID
 	metadataLeader.AddBroker(broker.Addr(), broker.BrokerID())
 	metadataLeader.AddTopic("test-topic", ErrNoError)
@@ -2067,7 +2067,7 @@ func TestTxnProduceRecordWithAbort(t *testing.T) {
 	config.Net.MaxOpenRequests = 1
 
 	metadataLeader := new(MetadataResponse)
-	metadataLeader.Version = 1
+	metadataLeader.Version = 4
 	metadataLeader.ControllerID = broker.brokerID
 	metadataLeader.AddBroker(broker.Addr(), broker.BrokerID())
 	metadataLeader.AddTopic("test-topic", ErrNoError)
@@ -2147,7 +2147,7 @@ func TestTxnCanAbort(t *testing.T) {
 	config.Net.MaxOpenRequests = 1
 
 	metadataLeader := new(MetadataResponse)
-	metadataLeader.Version = 1
+	metadataLeader.Version = 4
 	metadataLeader.ControllerID = broker.brokerID
 	metadataLeader.AddBroker(broker.Addr(), broker.BrokerID())
 	metadataLeader.AddTopic("test-topic", ErrNoError)

client_test.go

@@ -1071,7 +1071,7 @@ func TestClientCoordinatorConnectionRefused(t *testing.T) {
 func TestInitProducerIDConnectionRefused(t *testing.T) {
 	t.Parallel()
 	seedBroker := NewMockBroker(t, 1)
-	seedBroker.Returns(&MetadataResponse{Version: 1})
+	seedBroker.Returns(&MetadataResponse{Version: 4})
 
 	config := NewTestConfig()
 	config.Producer.Idempotent = true

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
 	}
 }