Producer.Retry.BackoffFunc proof of concept

async_producer.go

@@ -440,7 +440,14 @@ func (pp *partitionProducer) dispatch() {
 		if msg.retries > pp.highWatermark {
 			// a new, higher, retry level; handle it and then back off
 			pp.newHighWatermark(msg.retries)
-			time.Sleep(pp.parent.conf.Producer.Retry.Backoff)
+			var backoff time.Duration
+			if pp.parent.conf.Producer.Retry.BackoffFunc != nil {
+				maxRetries := pp.parent.conf.Producer.Retry.Max
+				backoff = pp.parent.conf.Producer.Retry.BackoffFunc(msg.retries, maxRetries)
+			} else {
+				backoff = pp.parent.conf.Producer.Retry.Backoff
+			}
+			time.Sleep(backoff)
 		} else if pp.highWatermark > 0 {
 			// we are retrying something (else highWatermark would be 0) but this message is not a *new* retry level
 			if msg.retries < pp.highWatermark {

config.go

@@ -168,6 +168,10 @@ type Config struct {
 			// (default 100ms). Similar to the `retry.backoff.ms` setting of the
 			// JVM producer.
 			Backoff time.Duration
+			// Called to compute backoff time dynamically. Useful for implementing
+			// more sophisticated backoff strategies. This takes precedence over
+			// `Backoff` if set.
+			BackoffFunc func(retries, maxRetries int) time.Duration
 		}
 	}