Merge pull request #2280 from Shopify/deps/git.luolix.top-klauspost-compr…

…ess-1.15.x

chore(deps): bump module github.com/klauspost/compress to v1.15.8

functional_consumer_test.go

@@ -438,50 +438,60 @@ func versionRange(lower KafkaVersion) []KafkaVersion {
 }
 
 func produceMsgs(t *testing.T, clientVersions []KafkaVersion, codecs []CompressionCodec, flush int, countPerVerCodec int, idempotent bool) []*ProducerMessage {
-	var wg sync.WaitGroup
-	var producedMessagesMu sync.Mutex
-	var producedMessages []*ProducerMessage
+	var (
+		wg                 sync.WaitGroup
+		producers          []SyncProducer
+		producedMessagesMu sync.Mutex
+		producedMessages   []*ProducerMessage
+	)
 	for _, prodVer := range clientVersions {
 		for _, codec := range codecs {
-			prodCfg := NewTestConfig()
-			prodCfg.Version = prodVer
-			prodCfg.Producer.Return.Successes = true
-			prodCfg.Producer.Return.Errors = true
-			prodCfg.Producer.Flush.MaxMessages = flush
-			prodCfg.Producer.Compression = codec
-			prodCfg.Producer.Idempotent = idempotent
-			if idempotent {
-				prodCfg.Producer.RequiredAcks = WaitForAll
-				prodCfg.Net.MaxOpenRequests = 1
-			}
+			t.Run("producer-"+prodVer.String()+"-"+codec.String(), func(t *testing.T) {
+				t.Logf("*** Producing with client version %s codec %s\n", prodVer, codec)
+				prodCfg := NewTestConfig()
+				prodCfg.Version = prodVer
+				prodCfg.Producer.Return.Successes = true
+				prodCfg.Producer.Return.Errors = true
+				prodCfg.Producer.Flush.MaxMessages = flush
+				prodCfg.Producer.Compression = codec
+				prodCfg.Producer.Idempotent = idempotent
+				if idempotent {
+					prodCfg.Producer.RequiredAcks = WaitForAll
+					prodCfg.Net.MaxOpenRequests = 1
+				}
 
-			p, err := NewSyncProducer(FunctionalTestEnv.KafkaBrokerAddrs, prodCfg)
-			if err != nil {
-				t.Errorf("Failed to create producer: version=%s, compression=%s, err=%v", prodVer, codec, err)
-				continue
-			}
-			defer safeClose(t, p)
-			for i := 0; i < countPerVerCodec; i++ {
-				msg := &ProducerMessage{
-					Topic: "test.1",
-					Value: StringEncoder(fmt.Sprintf("msg:%s:%s:%d", prodVer, codec, i)),
+				p, err := NewSyncProducer(FunctionalTestEnv.KafkaBrokerAddrs, prodCfg)
+				if err != nil {
+					t.Errorf("Failed to create producer: version=%s, compression=%s, err=%v", prodVer, codec, err)
+					return
 				}
-				wg.Add(1)
-				go func() {
-					defer wg.Done()
-					_, _, err := p.SendMessage(msg)
-					if err != nil {
-						t.Errorf("Failed to produce message: %s, err=%v", msg.Value, err)
+				producers = append(producers, p)
+				for i := 0; i < countPerVerCodec; i++ {
+					msg := &ProducerMessage{
+						Topic: "test.1",
+						Value: StringEncoder(fmt.Sprintf("msg:%s:%s:%d", prodVer, codec, i)),
 					}
-					producedMessagesMu.Lock()
-					producedMessages = append(producedMessages, msg)
-					producedMessagesMu.Unlock()
-				}()
-			}
+					wg.Add(1)
+					go func() {
+						defer wg.Done()
+						_, _, err := p.SendMessage(msg)
+						if err != nil {
+							t.Errorf("Failed to produce message: %s, err=%v", msg.Value, err)
+						}
+						producedMessagesMu.Lock()
+						producedMessages = append(producedMessages, msg)
+						producedMessagesMu.Unlock()
+					}()
+				}
+			})
 		}
 	}
 	wg.Wait()
 
+	for _, p := range producers {
+		safeClose(t, p)
+	}
+
 	// Sort produced message in ascending offset order.
 	sort.Slice(producedMessages, func(i, j int) bool {
 		return producedMessages[i].Offset < producedMessages[j].Offset
@@ -495,7 +505,7 @@ func consumeMsgs(t *testing.T, clientVersions []KafkaVersion, producedMessages [
 	// Consume all produced messages with all client versions supported by the
 	// cluster.
 	for _, consVer := range clientVersions {
-		t.Run(consVer.String(), func(t *testing.T) {
+		t.Run("consumer-"+consVer.String(), func(t *testing.T) {
 			t.Logf("*** Consuming with client version %s\n", consVer)
 			// Create a partition consumer that should start from the first produced
 			// message.

go.mod

@@ -13,7 +13,7 @@ require (
 	github.com/hashicorp/go-multierror v1.1.1
 	github.com/jcmturner/gofork v1.0.0
 	github.com/jcmturner/gokrb5/v8 v8.4.2
-	github.com/klauspost/compress v1.15.6
+	github.com/klauspost/compress v1.15.8
 	github.com/kr/pretty v0.3.0 // indirect
 	github.com/pierrec/lz4/v4 v4.1.15
 	github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475

go.sum

@@ -166,8 +166,8 @@ github.com/jstemmer/go-junit-report v0.9.1/go.mod h1:Brl9GWCQeLvo8nXZwPNNblvFj/X
 github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w=
 github.com/julienschmidt/httprouter v1.3.0/go.mod h1:JR6WtHb+2LUe8TCKY3cZOxFyyO8IZAc4RVcycCCAKdM=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
-github.com/klauspost/compress v1.15.6 h1:6D9PcO8QWu0JyaQ2zUMmu16T1T+zjjEpP91guRsvDfY=
-github.com/klauspost/compress v1.15.6/go.mod h1:PhcZ0MbTNciWF3rruxRgKxI5NkcHHrHUDtV4Yw2GlzU=
+github.com/klauspost/compress v1.15.8 h1:JahtItbkWjf2jzm/T+qgMxkP9EMHsqEUA6vCMGmXvhA=
+github.com/klauspost/compress v1.15.8/go.mod h1:PhcZ0MbTNciWF3rruxRgKxI5NkcHHrHUDtV4Yw2GlzU=
 github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/konsorten/go-windows-terminal-sequences v1.0.3/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc=