config: add producer option to configure transactions backoff duration

pkg/kgo/config.go

@@ -115,6 +115,7 @@ type cfg struct {
 	linger              time.Duration
 	recordTimeout       time.Duration
 	manualFlushing      bool
+	txnBackoff          time.Duration
 
 	partitioner Partitioner
 
@@ -460,6 +461,7 @@ func defaultCfg() cfg {
 		produceTimeout:      10 * time.Second,
 		recordRetries:       math.MaxInt64,             // effectively unbounded
 		partitioner:         StickyKeyPartitioner(nil), // default to how Kafka partitions
+		txnBackoff:          20 * time.Millisecond,
 
 		//////////////
 		// consumer //
@@ -780,6 +782,19 @@ func WithHooks(hooks ...Hook) Opt {
 	return clientOpt{func(cfg *cfg) { cfg.hooks = append(cfg.hooks, hooks...) }}
 }
 
+// ConcurrentTransactionBackoff sets the backoff interval to use during
+// transactional requests in case we encounter CONCURRENT_TRANSACTIONS error,
+// overriding the default 20ms.
+//
+// Sometimes, when a client begins a transaction quickly enough after finishing
+// a previous one, Kafka will return a CONCURRENT_TRANSACTIONS error. Clients
+// are expected to backoff slightly and retry the operation. Lower backoffs may
+// increase load on the brokers, while higher backoffs may increase transaction
+// latency in clients.
+func ConcurrentTransactionBackoff(backoff time.Duration) Opt {
+	return clientOpt{func(cfg *cfg) { cfg.txnBackoff = backoff }}
+}
+
 ////////////////////////////
 // PRODUCER CONFIGURATION //
 ////////////////////////////

pkg/kgo/txn.go

@@ -631,17 +631,18 @@ func (cl *Client) maybeRecoverProducerID() (necessary, did bool, err error) {
 func (cl *Client) doWithConcurrentTransactions(name string, fn func() error) error {
 	start := time.Now()
 	tries := 0
+	backoff := cl.cfg.txnBackoff
 start:
 	err := fn()
 	if err == kerr.ConcurrentTransactions && time.Since(start) < 10*time.Second {
 		tries++
 		cl.cfg.logger.Log(LogLevelInfo, fmt.Sprintf("%s failed with CONCURRENT_TRANSACTIONS, which may be because we ended a txn and began producing in a new txn too quickly; backing off and retrying", name),
-			"backoff", 100*time.Millisecond,
+			"backoff", backoff,
 			"since_request_tries_start", time.Since(start),
 			"tries", tries,
 		)
 		select {
-		case <-time.After(100 * time.Millisecond):
+		case <-time.After(backoff):
 		case <-cl.ctx.Done():
 			cl.cfg.logger.Log(LogLevelError, fmt.Sprintf("abandoning %s retry due to client ctx quitting", name))
 			return err