sink: tighten canFailFromLoadErrs

twmb · twmb · commit d620765d963b · 2022-11-30T21:45:44.000-07:00
The old version set canFailFromLoadErrs=false immediately once a batch was added to a request, well before the request is written. It is possible to have failures and delays that cause the request to never be written, but our record could never be failed. We now only set canFailFromLoadErrs=false at the moment we are about to write the request. Closes #239.
diff --git a/pkg/kgo/sink.go b/pkg/kgo/sink.go
@@ -1609,7 +1609,6 @@ func (p *produceRequest) tryAddBatch(produceVersion int32, recBuf *recBuf, batch
 	}
 
 	batch.tries++
-	batch.canFailFromLoadErrs = false
 	p.wireLength += batchWireLength
 	p.batches.addBatch(
 		recBuf.topic,
@@ -1932,6 +1931,7 @@ func (p *produceRequest) AppendTo(dst []byte) []byte {
 				batch.mu.Unlock()
 				continue
 			}
+			batch.canFailFromLoadErrs = false // we are going to write this batch: the response status is now unknown
 			var pmetrics ProduceBatchMetrics
 			if p.version < 3 {
 				dst, pmetrics = batch.appendToAsMessageSet(dst, uint8(p.version), p.compressor)

Original file line number	Diff line number	Diff line change
`@@ -1609,7 +1609,6 @@ func (p produceRequest) tryAddBatch(produceVersion int32, recBuf recBuf, batch`
`1609`	`1609`	`}`
`1610`	`1610`
`1611`	`1611`	`batch.tries++`
`1612`		`- batch.canFailFromLoadErrs = false`
`1613`	`1612`	`p.wireLength += batchWireLength`
`1614`	`1613`	`p.batches.addBatch(`
`1615`	`1614`	`recBuf.topic,`
`@@ -1932,6 +1931,7 @@ func (p *produceRequest) AppendTo(dst []byte) []byte {`
`1932`	`1931`	`batch.mu.Unlock()`
`1933`	`1932`	`continue`
`1934`	`1933`	`}`
	`1934`	`+ batch.canFailFromLoadErrs = false // we are going to write this batch: the response status is now unknown`
`1935`	`1935`	`var pmetrics ProduceBatchMetrics`
`1936`	`1936`	`if p.version < 3 {`
`1937`	`1937`	`dst, pmetrics = batch.appendToAsMessageSet(dst, uint8(p.version), p.compressor)`