Merge pull request #787 from twmb/777

kgo: fix deadlock in Produce when using MaxBufferedBytes

.github/workflows/lint-and-test.yml

@@ -31,7 +31,6 @@ jobs:
 
   integration-test-kafka:
     if: github.repository == 'twmb/franz-go'
-    needs: golangci
     runs-on: ubuntu-latest
     name: "integration test kafka"
     container: golang:latest
@@ -55,15 +54,14 @@ jobs:
           KAFKA_KRAFT_CLUSTER_ID: XkpGZQ27R3eTl3OdTm2LYA # 16 byte base64-encoded UUID
     steps:
       - uses: actions/checkout@v4
-      - run: go test ./...
+      - run: go test -timeout 5m ./...
         env:
           KGO_TEST_RF: 1
           KGO_SEEDS: kafka:9092
           KGO_TEST_STABLE_FETCH: true
 
   integration-test-redpanda:
     if: github.repository == 'twmb/franz-go'
-    needs: golangci
     runs-on: ubuntu-latest
     name: "integration test redpanda"
     container: golang:latest
@@ -76,7 +74,7 @@ jobs:
           REDPANDA_ADVERTISE_KAFKA_ADDRESS: redpanda:9092
     steps:
       - uses: actions/checkout@v4
-      - run: go test ./...
+      - run: go test -timeout 5m ./...
         env:
           KGO_TEST_RF: 1
           KGO_SEEDS: redpanda:9092

pkg/kgo/consumer_direct_test.go

@@ -338,7 +338,8 @@ func TestPauseIssue489(t *testing.T) {
 					exit.Store(true)
 				}
 			})
-			time.Sleep(100 * time.Microsecond)
+			cl.Flush(ctx)
+			time.Sleep(50 * time.Microsecond)
 		}
 	}()
 	defer cancel()
@@ -416,7 +417,8 @@ func TestPauseIssueOct2023(t *testing.T) {
 					exit.Store(true)
 				}
 			})
-			time.Sleep(100 * time.Microsecond)
+			cl.Flush(ctx)
+			time.Sleep(50 * time.Microsecond)
 		}
 	}()
 	defer cancel()

pkg/kgo/produce_request_test.go

@@ -2,13 +2,75 @@ package kgo
 
 import (
 	"bytes"
+	"context"
+	"errors"
 	"hash/crc32"
+	"math/rand"
+	"strings"
+	"sync"
+	"sync/atomic"
 	"testing"
 
 	"github.com/twmb/franz-go/pkg/kbin"
 	"github.com/twmb/franz-go/pkg/kmsg"
 )
 
+func TestClient_Produce(t *testing.T) {
+	var (
+		topic, cleanup = tmpTopicPartitions(t, 1)
+		numWorkers     = 50
+		recsToWrite    = int64(20_000)
+
+		workers      sync.WaitGroup
+		writeSuccess atomic.Int64
+		writeFailure atomic.Int64
+
+		randRec = func() *Record {
+			return &Record{
+				Key:   []byte("test"),
+				Value: []byte(strings.Repeat("x", rand.Intn(1000))),
+				Topic: topic,
+			}
+		}
+	)
+	defer cleanup()
+
+	cl, _ := newTestClient(MaxBufferedBytes(5000))
+	defer cl.Close()
+
+	// Start N workers that will concurrently write to the same partition.
+	var recsWritten atomic.Int64
+	var fatal atomic.Bool
+	for i := 0; i < numWorkers; i++ {
+		workers.Add(1)
+
+		go func() {
+			defer workers.Done()
+
+			for recsWritten.Add(1) <= recsToWrite {
+				res := cl.ProduceSync(context.Background(), randRec())
+				if err := res.FirstErr(); err == nil {
+					writeSuccess.Add(1)
+				} else {
+					if !errors.Is(err, ErrMaxBuffered) {
+						t.Errorf("unexpected error: %v", err)
+						fatal.Store(true)
+					}
+
+					writeFailure.Add(1)
+				}
+			}
+		}()
+	}
+	workers.Wait()
+
+	t.Logf("writes succeeded: %d", writeSuccess.Load())
+	t.Logf("writes failed:    %d", writeFailure.Load())
+	if fatal.Load() {
+		t.Fatal("failed")
+	}
+}
+
 // This file contains golden tests against kmsg AppendTo's to ensure our custom
 // encoding is correct.