Repository: Shopify/sarama
Pull request #2182 (2022-03-16 10:37:38)

producer: ensure that the management message (fin) is never "leaked"
Since async producer now support multiple inflight messages
thanks to https://github.com/Shopify/sarama/pull/1686 and https://github.com/Shopify/sarama/pull/2094, it now may "leak"
the "fin" internal management message to Kafka (and to the client)
when broker producer is reconnecting to Kafka broker and retries
multiple inflight messages at the same time.

Steps to reproduce: start async producer and restart Kafka broker (potentially multiple times). 
This trigger leader update and retires current broker producer.
There's a chance that newly created broker producer is used to retry an
inflight message so it is also marked as "dying" but it does not contain
any explicit errors.
This results in a "fin" message to be sent to Kafka and reported as successful
operation to the client.

Versions affected: `v1.31.0` and above.

Sarama logs of the event (with the patch applied):
```
[Sarama] 2022/03/16 09:54:31 producer/broker/10001 starting up
[Sarama] 2022/03/16 09:54:31 producer/broker/10001 state change to [open] on my-topic/26
[Sarama] 2022/03/16 09:54:31 producer/leader/my-topic/26 selected broker 10001
[Sarama] 2022/03/16 09:54:31 producer/leader/my-topic/26 state change to [flushing-1]
[Sarama] 2022/03/16 09:54:31 producer/leader/my-topic/26 state change to [normal]
[Sarama] 2022/03/16 09:54:31 producer/leader/my-topic/26 state change to [retrying-1]
[Sarama] 2022/03/16 09:54:31 producer/broker/10001 state change to [dying-1] on my-topic/26
[Sarama] 2022/03/16 09:54:31 producer/leader/my-topic/26 abandoning broker 10001
[Sarama] 2022/03/16 09:54:31 producer/broker/10001 input chan closed
[Sarama] 2022/03/16 09:54:33 producer/broker/10001 state change to [open] on my-topic/26
[Sarama] 2022/03/16 09:54:33 producer/leader/my-topic/26 selected broker 10001
[Sarama] 2022/03/16 09:54:33 producer/leader/my-topic/26 state change to [flushing-1]
[Sarama] 2022/03/16 09:54:33 producer/leader/my-topic/26 state change to [normal]
```

Kafka version: `2.7.1`
Go version: `1.17.7`

Kafka producer config:
```
cfg.Producer.Retry.Backoff = 2 * time.Second
cfg.Producer.Return.Successes = true
cfg.Producer.Return.Errors = true
cfg.Producer.Flush.Messages = 1000
cfg.Producer.Flush.Frequency = 100 * time.Millisecond
cfg.ChannelBufferSize = 1024
```


Patch:
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 355876a..44792a0 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -13,7 +13,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        go-version: [1.16.x, 1.17.x]
+        go-version: [1.17.x, 1.18.x]
         kafka-version: [2.8.1, 3.0.0, 3.1.0]
         platform: [ubuntu-latest]
 
@@ -26,7 +26,7 @@ jobs:
     - uses: actions/checkout@v2
 
     - name: Setup Go
-      uses: actions/setup-go@v2
+      uses: actions/setup-go@v3
       with:
         go-version: ${{ matrix.go-version }}
 
@@ -48,16 +48,16 @@ jobs:
 
     - name: Lint
       uses: golangci/golangci-lint-action@v2
-      if: ${{ matrix.go-version == '1.17.x' }}
+      if: ${{ matrix.go-version == '1.18.x' }}
       env:
         GOFLAGS: -tags=functional
       with:
-        version: v1.44.0
+        version: v1.45.0
 
     - name: Test (Unit)
       run: make test
-      if: ${{ matrix.go-version != '1.17.x' }}
+      if: ${{ matrix.go-version != '1.18.x' }}
 
     - name: Test (Functional)
       run: make test_functional
-      if: ${{ matrix.go-version == '1.17.x' }}
+      if: ${{ matrix.go-version == '1.18.x' }}
diff --git a/.github/workflows/codeql-analysis.yml b/.github/workflows/codeql-analysis.yml
index 0f8a38e..3b35d3e 100644
--- a/.github/workflows/codeql-analysis.yml
+++ b/.github/workflows/codeql-analysis.yml
@@ -31,10 +31,10 @@ jobs:
       with:
         languages: ${{ matrix.language }}
 
-    - name: Set up Go 1.16
-      uses: actions/setup-go@v2
+    - name: Setup Go
+      uses: actions/setup-go@v3
       with:
-        go-version: 1.16
+        go-version: 1.17
 
     - name: Autobuild
       uses: github/codeql-action/autobuild@v1
diff --git a/.github/workflows/fuzz.yml b/.github/workflows/fuzz.yml
index a9197fc..02be608 100644
--- a/.github/workflows/fuzz.yml
+++ b/.github/workflows/fuzz.yml
@@ -18,9 +18,9 @@ jobs:
     - uses: actions/checkout@v2
 
     - name: Setup Go
-      uses: actions/setup-go@v2
+      uses: actions/setup-go@v3
       with:
-        go-version: 1.17.x
+        go-version: 1.18.x
 
     - name: Get Go environment
       id: go-env
@@ -38,8 +38,5 @@ jobs:
         restore-keys: |
           fuzz-go-
 
-    - name: Build Go tip
-      run: go install golang.org/dl/gotip@latest && gotip download
-
     - name: Run any fuzzing tests
-      run: gotip test -v -run=^Fuzz -test.fuzztime=5m ./...
+      run: go test -v -run=^Fuzz -test.fuzztime=5m ./...
diff --git a/async_producer.go b/async_producer.go
index 27398bc..343d3de 100644
--- a/async_producer.go
+++ b/async_producer.go
@@ -840,6 +840,14 @@ func (bp *brokerProducer) run() {
 				continue
 			}
 
+			if msg.flags&fin == fin {
+				// New broker producer that was caught up by the retry loop
+				bp.parent.retryMessage(msg, ErrShuttingDown)
+				DebugLogger.Printf("producer/broker/%d state change to [dying-%d] on %s/%d\n",
+					bp.broker.ID(), msg.retries, msg.Topic, msg.Partition)
+				continue
+			}
+
 			if bp.buffer.wouldOverflow(msg) {
 				Logger.Printf("producer/broker/%d maximum request accumulated, waiting for space\n", bp.broker.ID())
 				if err := bp.waitForSpace(msg, false); err != nil {
diff --git a/broker.go b/broker.go
index 2d75a8c..e481ad7 100644
--- a/broker.go
+++ b/broker.go
@@ -666,6 +666,17 @@ func (b *Broker) CreateAcls(request *CreateAclsRequest) (*CreateAclsResponse, er
 		return nil, err
 	}
 
+	errs := make([]error, 0)
+	for _, res := range response.AclCreationResponses {
+		if !errors.Is(res.Err, ErrNoError) {
+			errs = append(errs, res.Err)
+		}
+	}
+
+	if len(errs) > 0 {
+		return response, Wrap(ErrCreateACLs, errs...)
+	}
+
 	return response, nil
 }
 
diff --git a/errors.go b/errors.go
index 507002b..bb64090 100644
--- a/errors.go
+++ b/errors.go
@@ -64,6 +64,9 @@ var ErrReassignPartitions = errors.New("failed to reassign partitions for topic"
 // ErrDeleteRecords is the type of error returned when fail to delete the required records
 var ErrDeleteRecords = errors.New("kafka server: failed to delete records")
 
+// ErrCreateACLs is the type of error returned when ACL creation failed
+var ErrCreateACLs = errors.New("kafka server: failed to create one or more ACL rules")
+
 // MultiErrorFormat specifies the formatter applied to format multierrors. The
 // default implementation is a consensed version of the hashicorp/go-multierror
 // default one
diff --git a/message.go b/message.go
index fd0d1d9..c6f35a3 100644
--- a/message.go
+++ b/message.go
@@ -42,6 +42,28 @@ func (cc CompressionCodec) String() string {
 	}[int(cc)]
 }
 
+// UnmarshalText returns a CompressionCodec from its string representation.
+func (cc *CompressionCodec) UnmarshalText(text []byte) error {
+	codecs := map[string]CompressionCodec{
+		"none":   CompressionNone,
+		"gzip":   CompressionGZIP,
+		"snappy": CompressionSnappy,
+		"lz4":    CompressionLZ4,
+		"zstd":   CompressionZSTD,
+	}
+	codec, ok := codecs[string(text)]
+	if !ok {
+		return fmt.Errorf("cannot parse %q as a compression codec", string(text))
+	}
+	*cc = codec
+	return nil
+}
+
+// MarshalText transforms a CompressionCodec into its string representation.
+func (cc CompressionCodec) MarshalText() ([]byte, error) {
+	return []byte(cc.String()), nil
+}
+
 // Message is a kafka message type
 type Message struct {
 	Codec            CompressionCodec // codec used to compress the message contents
diff --git a/mockresponses.go b/mockresponses.go
index da81696..d26a448 100644
--- a/mockresponses.go
+++ b/mockresponses.go
@@ -998,6 +998,24 @@ func (mr *MockCreateAclsResponse) For(reqBody versionedDecoder) encoderWithHeade
 	return res
 }
 
+type MockCreateAclsResponseError struct {
+	t TestReporter
+}
+
+func NewMockCreateAclsResponseWithError(t TestReporter) *MockCreateAclsResponseError {
+	return &MockCreateAclsResponseError{t: t}
+}
+
+func (mr *MockCreateAclsResponseError) For(reqBody versionedDecoder) encoderWithHeader {
+	req := reqBody.(*CreateAclsRequest)
+	res := &CreateAclsResponse{}
+
+	for range req.AclCreations {
+		res.AclCreationResponses = append(res.AclCreationResponses, &AclCreationResponse{Err: ErrInvalidRequest})
+	}
+	return res
+}
+
 type MockListAclsResponse struct {
 	t TestReporter
 }


Test patch:
diff --git a/admin_test.go b/admin_test.go
index 3191fff..b0d1620 100644
--- a/admin_test.go
+++ b/admin_test.go
@@ -1086,6 +1086,38 @@ func TestClusterAdminCreateAcl(t *testing.T) {
 	}
 }
 
+func TestClusterAdminCreateAclErrorHandling(t *testing.T) {
+	seedBroker := NewMockBroker(t, 1)
+	defer seedBroker.Close()
+
+	seedBroker.SetHandlerByMap(map[string]MockResponse{
+		"MetadataRequest": NewMockMetadataResponse(t).
+			SetController(seedBroker.BrokerID()).
+			SetBroker(seedBroker.Addr(), seedBroker.BrokerID()),
+		"CreateAclsRequest": NewMockCreateAclsResponseWithError(t),
+	})
+
+	config := NewTestConfig()
+	config.Version = V1_0_0_0
+	admin, err := NewClusterAdmin([]string{seedBroker.Addr()}, config)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	r := Resource{ResourceType: AclResourceTopic, ResourceName: "my_topic"}
+	a := Acl{Host: "localhost", Operation: AclOperationAlter, PermissionType: AclPermissionAny}
+
+	err = admin.CreateACL(r, a)
+	if err == nil {
+		t.Fatal(errors.New("error should have been thrown"))
+	}
+
+	err = admin.Close()
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
 func TestClusterAdminListAcls(t *testing.T) {
 	seedBroker := NewMockBroker(t, 1)
 	defer seedBroker.Close()
diff --git a/async_producer_test.go b/async_producer_test.go
index 3770966..c2ae69e 100644
--- a/async_producer_test.go
+++ b/async_producer_test.go
@@ -17,31 +17,67 @@ import (
 
 const TestMessage = "ABC THE MESSAGE"
 
-func closeProducer(t *testing.T, p AsyncProducer) {
+func closeProducerWithTimeout(t *testing.T, p AsyncProducer, timeout time.Duration) {
 	var wg sync.WaitGroup
 	p.AsyncClose()
 
+	closer := make(chan struct{})
+	timer := time.AfterFunc(timeout, func() {
+		t.Error("timeout")
+		close(closer)
+	})
+	defer timer.Stop()
+
 	wg.Add(2)
 	go func() {
-		for range p.Successes() {
-			t.Error("Unexpected message on Successes()")
+		defer wg.Done()
+		for {
+			select {
+			case <-closer:
+				return
+			case _, ok := <-p.Successes():
+				if !ok {
+					return
+				}
+				t.Error("Unexpected message on Successes()")
+			}
 		}
-		wg.Done()
 	}()
 	go func() {
-		for msg := range p.Errors() {
-			t.Error(msg.Err)
+		defer wg.Done()
+		for {
+			select {
+			case <-closer:
+				return
+			case msg, ok := <-p.Errors():
+				if !ok {
+					return
+				}
+				t.Error(msg.Err)
+			}
 		}
-		wg.Done()
 	}()
 	wg.Wait()
 }
 
-func expectResults(t *testing.T, p AsyncProducer, successes, errors int) {
+func closeProducer(t *testing.T, p AsyncProducer) {
+	closeProducerWithTimeout(t, p, 5*time.Minute)
+}
+
+func expectResultsWithTimeout(t *testing.T, p AsyncProducer, successes, errors int, timeout time.Duration) {
 	t.Helper()
 	expect := successes + errors
+	defer func() {
+		if successes != 0 || errors != 0 {
+			t.Error("Unexpected successes", successes, "or errors", errors)
+		}
+	}()
+	timer := time.NewTimer(timeout)
+	defer timer.Stop()
 	for expect > 0 {
 		select {
+		case <-timer.C:
+			return
 		case msg := <-p.Errors():
 			if msg.Msg.flags != 0 {
 				t.Error("Message had flags set")
@@ -62,9 +98,10 @@ func expectResults(t *testing.T, p AsyncProducer, successes, errors int) {
 			}
 		}
 	}
-	if successes != 0 || errors != 0 {
-		t.Error("Unexpected successes", successes, "or errors", errors)
-	}
+}
+
+func expectResults(t *testing.T, p AsyncProducer, successes, errors int) {
+	expectResultsWithTimeout(t, p, successes, errors, 5*time.Minute)
 }
 
 type testPartitioner chan *int32
@@ -693,6 +730,112 @@ func TestAsyncProducerMultipleRetriesWithConcurrentRequests(t *testing.T) {
 	closeProducer(t, producer)
 }
 
+func TestAsyncProducerBrokerRestart(t *testing.T) {
+	// Logger = log.New(os.Stdout, "[sarama] ", log.LstdFlags)
+
+	seedBroker := NewMockBroker(t, 1)
+	leader := NewMockBroker(t, 2)
+
+	var leaderLock sync.Mutex
+
+	// The seed broker only handles Metadata request
+	seedBroker.setHandler(func(req *request) (res encoderWithHeader) {
+		leaderLock.Lock()
+		defer leaderLock.Unlock()
+		metadataLeader := new(MetadataResponse)
+		metadataLeader.AddBroker(leader.Addr(), leader.BrokerID())
+		metadataLeader.AddTopicPartition("my_topic", 0, leader.BrokerID(), nil, nil, nil, ErrNoError)
+		return metadataLeader
+	})
+
+	var emptyValues int32 = 0
+
+	countRecordsWithEmptyValue := func(req *request) {
+		preq := req.body.(*ProduceRequest)
+		if batch := preq.records["my_topic"][0].RecordBatch; batch != nil {
+			for _, record := range batch.Records {
+				if len(record.Value) == 0 {
+					atomic.AddInt32(&emptyValues, 1)
+				}
+			}
+		}
+		if batch := preq.records["my_topic"][0].MsgSet; batch != nil {
+			for _, record := range batch.Messages {
+				if len(record.Msg.Value) == 0 {
+					atomic.AddInt32(&emptyValues, 1)
+				}
+			}
+		}
+	}
+
+	leader.setHandler(func(req *request) (res encoderWithHeader) {
+		countRecordsWithEmptyValue(req)
+
+		time.Sleep(50 * time.Millisecond)
+
+		prodSuccess := new(ProduceResponse)
+		prodSuccess.AddTopicPartition("my_topic", 0, ErrNotLeaderForPartition)
+		return prodSuccess
+	})
+
+	config := NewTestConfig()
+	config.Producer.Retry.Backoff = 250 * time.Millisecond
+	config.Producer.Flush.MaxMessages = 1
+	config.Producer.Return.Errors = true
+	config.Producer.Return.Successes = true
+	config.Producer.Retry.Max = 10
+
+	producer, err := NewAsyncProducer([]string{seedBroker.Addr()}, config)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	var wg sync.WaitGroup
+
+	pushMsg := func() {
+		defer wg.Done()
+		for i := 0; i < 10; i++ {
+			producer.Input() <- &ProducerMessage{Topic: "my_topic", Key: nil, Value: StringEncoder(TestMessage)}
+			time.Sleep(50 * time.Millisecond)
+		}
+	}
+
+	wg.Add(1)
+	go pushMsg()
+
+	for i := 0; i < 3; i++ {
+		time.Sleep(100 * time.Millisecond)
+
+		wg.Add(1)
+		go pushMsg()
+	}
+
+	leader.Close()
+	leaderLock.Lock()
+	leader = NewMockBroker(t, 2)
+	leaderLock.Unlock()
+	leader.setHandler(func(req *request) (res encoderWithHeader) {
+		countRecordsWithEmptyValue(req)
+
+		prodSuccess := new(ProduceResponse)
+		prodSuccess.AddTopicPartition("my_topic", 0, ErrNoError)
+		return prodSuccess
+	})
+
+	wg.Wait()
+
+	expectResultsWithTimeout(t, producer, 40, 0, 10*time.Second)
+
+	seedBroker.Close()
+	leader.Close()
+
+	closeProducerWithTimeout(t, producer, 5*time.Second)
+
+	if emptyValues := atomic.LoadInt32(&emptyValues); emptyValues > 0 {
+		t.Fatalf("%d empty values", emptyValues)
+	}
+}
+
 func TestAsyncProducerOutOfRetries(t *testing.T) {
 	t.Skip("Enable once bug #294 is fixed.")
 
diff --git a/message_test.go b/message_test.go
index a6c7cff..d7bd430 100644
--- a/message_test.go
+++ b/message_test.go
@@ -244,3 +244,32 @@ func TestMessageDecodingUnknownVersions(t *testing.T) {
 		t.Error("Decoding an unknown magic byte produced an unknown error ", err)
 	}
 }
+
+func TestCompressionCodecUnmarshal(t *testing.T) {
+	cases := []struct {
+		Input         string
+		Expected      CompressionCodec
+		ExpectedError bool
+	}{
+		{"none", CompressionNone, false},
+		{"zstd", CompressionZSTD, false},
+		{"gzip", CompressionGZIP, false},
+		{"unknown", CompressionNone, true},
+	}
+	for _, c := range cases {
+		var cc CompressionCodec
+		err := cc.UnmarshalText([]byte(c.Input))
+		if err != nil && !c.ExpectedError {
+			t.Errorf("UnmarshalText(%q) error:\n%+v", c.Input, err)
+			continue
+		}
+		if err == nil && c.ExpectedError {
+			t.Errorf("UnmarshalText(%q) got %v but expected error", c.Input, cc)
+			continue
+		}
+		if cc != c.Expected {
+			t.Errorf("UnmarshalText(%q) got %v but expected %v", c.Input, cc, c.Expected)
+			continue
+		}
+	}
+}