Merge pull request #846 from Shopify/4-0-fix

Use sarama version to build the correct OffsetRequest call

Author: bai

core/internal/cluster/kafka_cluster.go

@@ -227,6 +227,22 @@ func (module *KafkaCluster) generateOffsetRequests(client helpers.SaramaClient)
 			}
 			if _, ok := requests[broker.ID()]; !ok {
 				requests[broker.ID()] = &sarama.OffsetRequest{}
+				// Match the version of the client as sarama's getOffset function does
+				// https://github.com/IBM/sarama/blob/main/client.go#L863-L876
+				if client.Config().Version.IsAtLeast(sarama.V2_1_0_0) {
+					// Version 4 adds the current leader epoch, which is used for fencing.
+					requests[broker.ID()].Version = 4
+				} else if client.Config().Version.IsAtLeast(sarama.V2_0_0_0) {
+					// Version 3 is the same as version 2.
+					requests[broker.ID()].Version = 3
+				} else if client.Config().Version.IsAtLeast(sarama.V0_11_0_0) {
+					// Version 2 adds the isolation level, which is used for transactional reads.
+					requests[broker.ID()].Version = 2
+				} else if client.Config().Version.IsAtLeast(sarama.V0_10_1_0) {
+					// Version 1 removes MaxNumOffsets.  From this version forward, only a single
+					// offset can be returned.
+					requests[broker.ID()].Version = 1
+				}
 			}
 			brokers[broker.ID()] = broker
 			requests[broker.ID()].AddBlock(topic, partitionID, sarama.OffsetNewest, 1)

core/internal/cluster/kafka_cluster_test.go

@@ -172,6 +172,7 @@ func TestKafkaCluster_generateOffsetRequests(t *testing.T) {
 	// Set up the mock to return the leader broker for a test topic and partition
 	client := &helpers.MockSaramaClient{}
 	client.On("Leader", "testtopic", int32(0)).Return(broker, nil)
+	client.On("Config").Return(sarama.NewConfig())
 
 	requests, brokers := module.generateOffsetRequests(client)
 
@@ -199,6 +200,7 @@ func TestKafkaCluster_generateOffsetRequests_NoLeader(t *testing.T) {
 	var nilBroker *helpers.BurrowSaramaBroker
 	client.On("Leader", "testtopic", int32(0)).Return(nilBroker, errors.New("no leader error"))
 	client.On("Leader", "testtopic", int32(1)).Return(broker, nil)
+	client.On("Config").Return(sarama.NewConfig())
 
 	requests, brokers := module.generateOffsetRequests(client)
 
@@ -233,6 +235,7 @@ func TestKafkaCluster_getOffsets(t *testing.T) {
 	var nilBroker *helpers.BurrowSaramaBroker
 	client.On("Leader", "testtopic", int32(0)).Return(broker, nil)
 	client.On("Leader", "testtopic", int32(1)).Return(nilBroker, errors.New("no leader error"))
+	client.On("Config").Return(sarama.NewConfig())
 
 	go module.getOffsets(client)
 	request := <-module.App.StorageChannel
@@ -274,7 +277,7 @@ func TestKafkaCluster_getOffsets_BrokerFailed(t *testing.T) {
 	// Set up the mock to return the leader broker for a test topic and partition
 	client := &helpers.MockSaramaClient{}
 	client.On("Leader", "testtopic", int32(0)).Return(broker, nil)
-
+	client.On("Config").Return(sarama.NewConfig())
 	module.getOffsets(client)
 
 	broker.AssertExpectations(t)