concurrent broker topic offset fetches

pkg/scalers/kafka_scaler.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"strconv"
 	"strings"
+	"sync"
 
 	"github.com/Shopify/sarama"
 	v2beta2 "k8s.io/api/autoscaling/v2beta2"
@@ -406,6 +407,11 @@ func (s *kafkaScaler) GetMetrics(ctx context.Context, metricName string, metricS
 	return append([]external_metrics.ExternalMetricValue{}, metric), nil
 }
 
+type brokerOffsetResult struct {
+	offsetResp *sarama.OffsetResponse
+	err 	   error
+}
+
 func (s *kafkaScaler) getTopicOffsets(partitions []int32) (map[int32]int64, error) {
 	version := int16(0)
 	if s.client.Config().Version.IsAtLeast(sarama.V0_10_1_0) {
@@ -430,26 +436,45 @@ func (s *kafkaScaler) getTopicOffsets(partitions []int32) (map[int32]int64, erro
 		request.AddBlock(s.metadata.topic, partitionID, sarama.OffsetNewest, 1)
 	}
 
-	offsets := make(map[int32]int64)
 
 	// Step 2: send requests, one per broker, and collect offsets
+	resultCh := make(chan brokerOffsetResult)
+	var wg sync.WaitGroup
+	wg.Add(len(requests))
 	for broker, request := range requests {
-		response, err := broker.GetAvailableOffsets(request)
+		go func() {
+			response, err := broker.GetAvailableOffsets(request)
+			resultCh <- brokerOffsetResult{response, err}
+			wg.Done()
+		}()
+	}
 
-		if err != nil {
-			return nil, err
+	go func() {
+		wg.Wait()
+		close(resultCh)
+	}()
+
+	var mtx = &sync.RWMutex{}
+	offsets := make(map[int32]int64)
+	for brokerOffsetRes := range resultCh {
+		if brokerOffsetRes.err != nil {
+			close(resultCh)
+			return nil, brokerOffsetRes.err
 		}
 
-		for _, blocks := range response.Blocks {
+		for _, blocks := range brokerOffsetRes.offsetResp.Blocks {
 			for partitionID, block := range blocks {
 				if block.Err != sarama.ErrNoError {
 					return nil, block.Err
 				}
 
+				mtx.Lock()
 				offsets[partitionID] = block.Offset
+				mtx.Unlock()
 			}
 		}
 	}
 
 	return offsets, nil
 }
+