Repository: influxdata/influxdb-client-go
Pull request #271 (2021-08-11 20:10:57)

feat: use exponential retry strategy
## Proposed Changes

This PR aligns retry strategy implementations across all official InfluxDB client libraries so that a delay for the next retry delay is a random value in the interval _retryInterval * exponentialBase^(attempts)_ and _retryInterval * exponentialBase^(attempts+1)_.
The defaults were changed:
- retryInterval=5_000
- exponentialBase=2 (it is now also configurable)
- maxRetryDelay=125_000
- maxRetries=5

Retry delays are by default randomly distributed within the ranges of [5_000-10_000, 10_000-20_000, 20_000-40_000, 40_000-80_000, 80_000-125_000]. 

Added is also _MaxRetryTime_ option. When an overall time spent by retrying exceeds a maxRetryTime (180_000 millis by default), the write is not retried and fails.

## Checklist

<!-- Checkboxes below this note can be erased if not applicable to your Pull Request. -->

- [X] CHANGELOG.md updated
- [X] Rebased/mergeable
- [X] A test has been added if appropriate
- [X] Tests pass
- [X] Commit messages are in [semantic format](https://seesparkbox.com/foundry/semantic_commit_messages)




Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index bbc14fb..0388315 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,6 +1,7 @@
 ## 2.5.0 [in progress]
 ### Features
  - [#264](https://github.com/influxdata/influxdb-client-go/pull/264) Synced generated server API with the latest [oss.yml](https://github.com/influxdata/openapi/blob/master/contracts/oss.yml). 
+ - [#271](https://github.com/influxdata/influxdb-client-go/pull/271) Use exponential _random_ retry strategy 
 
 ### Bug fixes
  - [#270](https://github.com/influxdata/influxdb-client-go/pull/270) Fixed duplicate `Content-Type` header in requests to managemet API  
diff --git a/api/write.go b/api/write.go
index 32a4122..f4890a5 100644
--- a/api/write.go
+++ b/api/write.go
@@ -146,7 +146,7 @@ x:
 func (w *WriteAPIImpl) flushBuffer() {
 	if len(w.writeBuffer) > 0 {
 		log.Info("sending batch")
-		batch := iwrite.NewBatch(buffer(w.writeBuffer), w.writeOptions.RetryInterval())
+		batch := iwrite.NewBatch(buffer(w.writeBuffer), w.writeOptions.RetryInterval(), w.writeOptions.MaxRetryTime())
 		w.writeCh <- batch
 		w.writeBuffer = w.writeBuffer[:0]
 	}
diff --git a/api/write/options.go b/api/write/options.go
index 8bd13c8..04680d2 100644
--- a/api/write/options.go
+++ b/api/write/options.go
@@ -21,14 +21,18 @@ type Options struct {
 	useGZip bool
 	// Tags added to each point during writing. If a point already has a tag with the same key, it is left unchanged.
 	defaultTags map[string]string
-	// Default retry interval in ms, if not sent by server. Default 5,000ms
+	// Default retry interval in ms, if not sent by server. Default 5,000.
 	retryInterval uint
-	// Maximum count of retry attempts of failed writes
+	// Maximum count of retry attempts of failed writes, default 5.
 	maxRetries uint
-	// Maximum number of points to keep for retry. Should be multiple of BatchSize. Default 50,000
+	// Maximum number of points to keep for retry. Should be multiple of BatchSize. Default 50,000.
 	retryBufferLimit uint
-	// Maximum retry interval, default 5min (300,000ms)
+	// The maximum delay between each retry attempt in milliseconds, default 125,000.
 	maxRetryInterval uint
+	// The maximum total retry timeout in millisecond, default 180,000.
+	maxRetryTime uint
+	// The base for the exponential retry delay
+	exponentialBase uint
 }
 
 // BatchSize returns size of batch
@@ -53,18 +57,18 @@ func (o *Options) SetFlushInterval(flushIntervalMs uint) *Options {
 	return o
 }
 
-// RetryInterval returns the retry interval in ms
+// RetryInterval returns the default retry interval in ms, if not sent by server. Default 5,000.
 func (o *Options) RetryInterval() uint {
 	return o.retryInterval
 }
 
-// SetRetryInterval sets retry interval in ms, which is set if not sent by server
+// SetRetryInterval sets the time to wait before retry unsuccessful write in ms, if not sent by server
 func (o *Options) SetRetryInterval(retryIntervalMs uint) *Options {
 	o.retryInterval = retryIntervalMs
 	return o
 }
 
-// MaxRetries returns maximum count of retry attempts of failed writes
+// MaxRetries returns maximum count of retry attempts of failed writes, default 5.
 func (o *Options) MaxRetries() uint {
 	return o.maxRetries
 }
@@ -76,7 +80,7 @@ func (o *Options) SetMaxRetries(maxRetries uint) *Options {
 	return o
 }
 
-// RetryBufferLimit returns retry buffer limit
+// RetryBufferLimit returns retry buffer limit.
 func (o *Options) RetryBufferLimit() uint {
 	return o.retryBufferLimit
 }
@@ -87,17 +91,39 @@ func (o *Options) SetRetryBufferLimit(retryBufferLimit uint) *Options {
 	return o
 }
 
-// MaxRetryInterval return maximum retry interval in ms. Default 5min.
+// MaxRetryInterval returns the maximum delay between each retry attempt in milliseconds, default 125,000.
 func (o *Options) MaxRetryInterval() uint {
 	return o.maxRetryInterval
 }
 
-// SetMaxRetryInterval set maximum retry interval in ms
+// SetMaxRetryInterval sets the maximum delay between each retry attempt in millisecond
 func (o *Options) SetMaxRetryInterval(maxRetryIntervalMs uint) *Options {
 	o.maxRetryInterval = maxRetryIntervalMs
 	return o
 }
 
+// MaxRetryTime returns the maximum total retry timeout in millisecond, default 180,000.
+func (o *Options) MaxRetryTime() uint {
+	return o.maxRetryTime
+}
+
+// SetMaxRetryTime sets the maximum total retry timeout in millisecond.
+func (o *Options) SetMaxRetryTime(maxRetryTimeMs uint) *Options {
+	o.maxRetryTime = maxRetryTimeMs
+	return o
+}
+
+// ExponentialBase returns the base for the exponential retry delay. Default 2.
+func (o *Options) ExponentialBase() uint {
+	return o.exponentialBase
+}
+
+// SetExponentialBase sets the base for the exponential retry delay.
+func (o *Options) SetExponentialBase(retryExponentialBase uint) *Options {
+	o.exponentialBase = retryExponentialBase
+	return o
+}
+
 // Precision returns time precision for writes
 func (o *Options) Precision() time.Duration {
 	return o.precision
@@ -138,5 +164,6 @@ func (o *Options) DefaultTags() map[string]string {
 
 // DefaultOptions returns Options object with default values
 func DefaultOptions() *Options {
-	return &Options{batchSize: 5000, maxRetries: 3, retryInterval: 5000, maxRetryInterval: 300000, flushInterval: 1000, precision: time.Nanosecond, useGZip: false, retryBufferLimit: 50000, defaultTags: make(map[string]string)}
+	return &Options{batchSize: 5_000, flushInterval: 1_000, precision: time.Nanosecond, useGZip: false, retryBufferLimit: 50_000, defaultTags: make(map[string]string),
+		maxRetries: 5, retryInterval: 5_000, maxRetryInterval: 125_000, maxRetryTime: 180_000, exponentialBase: 2}
 }
diff --git a/api/writeAPIBlocking.go b/api/writeAPIBlocking.go
index b11700f..3bd43f0 100644
--- a/api/writeAPIBlocking.go
+++ b/api/writeAPIBlocking.go
@@ -72,7 +72,7 @@ func NewWriteAPIBlocking(org string, bucket string, service http2.Service, write
 }
 
 func (w *writeAPIBlocking) write(ctx context.Context, line string) error {
-	err := w.service.WriteBatch(ctx, iwrite.NewBatch(line, w.writeOptions.RetryInterval()))
+	err := w.service.WriteBatch(ctx, iwrite.NewBatch(line, w.writeOptions.RetryInterval(), w.writeOptions.MaxRetryTime()))
 	if err != nil {
 		return err.Unwrap()
 	}
diff --git a/internal/write/service.go b/internal/write/service.go
index 029a026..5d6abc1 100644
--- a/internal/write/service.go
+++ b/internal/write/service.go
@@ -8,7 +8,9 @@ package write
 import (
 	"bytes"
 	"context"
+	"fmt"
 	"io"
+	"math/rand"
 	"net/http"
 	"net/url"
 	"sort"
@@ -26,31 +28,37 @@ import (
 
 // Batch holds information for sending points batch
 type Batch struct {
-	batch         string
-	retryDelay    uint
+	// lines to send
+	batch string
+	// current retry delay
+	retryDelay uint
+	// retry attempts so far
 	retryAttempts uint
-	evicted       bool
+	// true if it was removed from queue
+	evicted bool
+	// time where this batch expires
+	expires time.Time
 }
 
 // NewBatch creates new batch
-func NewBatch(data string, retryDelay uint) *Batch {
+func NewBatch(data string, retryDelay uint, expireDelayMs uint) *Batch {
 	return &Batch{
 		batch:      data,
 		retryDelay: retryDelay,
+		expires:    time.Now().Add(time.Duration(expireDelayMs) * time.Millisecond),
 	}
 }
 
 // Service is responsible for reliable writing of batches
 type Service struct {
-	org                  string
-	bucket               string
-	httpService          http2.Service
-	url                  string
-	lastWriteAttempt     time.Time
-	retryQueue           *queue
-	lock                 sync.Mutex
-	writeOptions         *write.Options
-	retryExponentialBase uint
+	org              string
+	bucket           string
+	httpService      http2.Service
+	url              string
+	lastWriteAttempt time.Time
+	retryQueue       *queue
+	lock             sync.Mutex
+	writeOptions     *write.Options
 }
 
 // NewService creates new write service
@@ -68,10 +76,19 @@ func NewService(org string, bucket string, httpService http2.Service, options *w
 	params.Set("precision", precisionToString(options.Precision()))
 	u.RawQuery = params.Encode()
 	writeURL := u.String()
-	return &Service{org: org, bucket: bucket, httpService: httpService, url: writeURL, writeOptions: options, retryQueue: newQueue(int(retryBufferLimit)), retryExponentialBase: 5}
+	return &Service{org: org, bucket: bucket, httpService: httpService, url: writeURL, writeOptions: options, retryQueue: newQueue(int(retryBufferLimit))}
 }
 
-// HandleWrite handles writes batches and handles retrying
+// HandleWrite handles writes of batches and handles retrying.
+// Retrying is triggered by new writes, there is no scheduler.
+// It first checks retry queue, cause it has highest priority.
+// If there are some batches in retry queue, those are written and incoming batch is added to end of retry queue.
+// Immediate write is allowed only in case there was success or not retryable error.
+// Otherwise delay is checked based on recent batch.
+// If write of batch fails with retryable error (connection errors and HTTP code >= 429),
+// Batch retry time is calculated based on #of attempts.
+// If writes continues failing and # of attempts reaches maximum or total retry time reaches maxRetryTime,
+// batch is discarded.
 func (w *Service) HandleWrite(ctx context.Context, batch *Batch) error {
 	log.Debug("Write proc: received write request")
 	batchToWrite := batch
@@ -111,6 +128,12 @@ func (w *Service) HandleWrite(ctx context.Context, batch *Batch) error {
 		}
 		// write batch
 		if batchToWrite != nil {
+			if time.Now().After(batchToWrite.expires) {
+				if !batchToWrite.evicted {
+					w.retryQueue.pop()
+				}
+				return fmt.Errorf("write failed (attempts %d): max retry time exceeded", batchToWrite.retryAttempts)
+			}
 			perror := w.WriteBatch(ctx, batchToWrite)
 			if perror != nil {
 				if w.writeOptions.MaxRetries() != 0 && (perror.StatusCode == 0 || perror.StatusCode >= http.StatusTooManyRequests) {
@@ -118,11 +141,7 @@ func (w *Service) HandleWrite(ctx context.Context, batch *Batch) error {
 					if perror.RetryAfter > 0 {
 						batchToWrite.retryDelay = perror.RetryAfter * 1000
 					} else {
-						exp := uint(1)
-						for i := uint(0); i < batchToWrite.retryAttempts; i++ {
-							exp = exp * w.retryExponentialBase
-						}
-						batchToWrite.retryDelay = min(w.writeOptions.RetryInterval()*exp, w.writeOptions.MaxRetryInterval())
+						batchToWrite.retryDelay = w.computeRetryDelay(batchToWrite.retryAttempts)
 					}
 					if batchToWrite.retryAttempts == 0 {
 						if w.retryQueue.push(batch) {
@@ -137,7 +156,7 @@ func (w *Service) HandleWrite(ctx context.Context, batch *Batch) error {
 				} else {
 					log.Errorf("Write error: %s\n", perror.Error())
 				}
-				return perror
+				return fmt.Errorf("write failed (attempts %d): %w", batchToWrite.retryAttempts, perror)
 			}
 			if retrying && !batchToWrite.evicted {
 				w.retryQueue.pop()
@@ -150,6 +169,31 @@ func (w *Service) HandleWrite(ctx context.Context, batch *Batch) error {
 	return nil
 }
 
+// computeRetryDelay calculates retry delay
+// Retry delay is calculated as random value within the interval
+// [retry_interval * exponential_base^(attempts) and retry_interval * exponential_base^(attempts+1)]
+func (w *Service) computeRetryDelay(attempts uint) uint {
+	minDelay := int(w.writeOptions.RetryInterval() * pow(w.writeOptions.ExponentialBase(), attempts))
+	maxDelay := int(w.writeOptions.RetryInterval() * pow(w.writeOptions.ExponentialBase(), attempts+1))
+	retryDelay := uint(rand.Intn(maxDelay-minDelay) + minDelay)
+	if retryDelay > w.writeOptions.MaxRetryInterval() {
+		retryDelay = w.writeOptions.MaxRetryInterval()
+	}
+	return retryDelay
+}
+
+// pow computes x**y
+func pow(x, y uint) uint {
+	p := uint(1)
+	if y == 0 {
+		return 1
+	}
+	for i := uint(1); i <= y; i++ {
+		p = p * x
+	}
+	return p
+}
+
 // WriteBatch performs actual writing via HTTP service
 func (w *Service) WriteBatch(ctx context.Context, batch *Batch) *http2.Error {
 	var body io.Reader
diff --git a/options.go b/options.go
index 43a3dff..000bfc2 100644
--- a/options.go
+++ b/options.go
@@ -56,7 +56,7 @@ func (o *Options) SetRetryInterval(retryIntervalMs uint) *Options {
 	return o
 }
 
-// MaxRetries returns maximum count of retry attempts of failed writes
+// MaxRetries returns maximum count of retry attempts of failed writes, default 5.
 func (o *Options) MaxRetries() uint {
 	return o.WriteOptions().MaxRetries()
 }
@@ -79,17 +79,39 @@ func (o *Options) SetRetryBufferLimit(retryBufferLimit uint) *Options {
 	return o
 }
 
-// MaxRetryInterval return maximum retry interval in ms. Default 5min.
+// MaxRetryInterval returns the maximum delay between each retry attempt in milliseconds, default 125,000.
 func (o *Options) MaxRetryInterval() uint {
 	return o.WriteOptions().MaxRetryInterval()
 }
 
-// SetMaxRetryInterval set maximum retry interval in ms
+// SetMaxRetryInterval sets the maximum delay between each retry attempt in millisecond.
 func (o *Options) SetMaxRetryInterval(maxRetryIntervalMs uint) *Options {
 	o.WriteOptions().SetMaxRetryInterval(maxRetryIntervalMs)
 	return o
 }
 
+// MaxRetryTime returns the maximum total retry timeout in millisecond, default 180,000.
+func (o *Options) MaxRetryTime() uint {
+	return o.WriteOptions().MaxRetryTime()
+}
+
+// SetMaxRetryTime sets the maximum total retry timeout in millisecond.
+func (o *Options) SetMaxRetryTime(maxRetryTimeMs uint) *Options {
+	o.WriteOptions().SetMaxRetryTime(maxRetryTimeMs)
+	return o
+}
+
+// ExponentialBase returns the base for the exponential retry delay. Default 2.
+func (o *Options) ExponentialBase() uint {
+	return o.WriteOptions().ExponentialBase()
+}
+
+// SetExponentialBase sets the base for the exponential retry delay.
+func (o *Options) SetExponentialBase(exponentialBase uint) *Options {
+	o.WriteOptions().SetExponentialBase(exponentialBase)
+	return o
+}
+
 // LogLevel returns log level
 func (o *Options) LogLevel() uint {
 	return o.logLevel


Test patch:
diff --git a/api/write/options_test.go b/api/write/options_test.go
index 91adad9..a30f18c 100644
--- a/api/write/options_test.go
+++ b/api/write/options_test.go
@@ -14,14 +14,16 @@ import (
 
 func TestDefaultOptions(t *testing.T) {
 	opts := write.DefaultOptions()
-	assert.Equal(t, uint(5000), opts.BatchSize())
-	assert.Equal(t, false, opts.UseGZip())
-	assert.Equal(t, uint(1000), opts.FlushInterval())
-	assert.Equal(t, time.Nanosecond, opts.Precision())
-	assert.Equal(t, uint(50000), opts.RetryBufferLimit())
-	assert.Equal(t, uint(5000), opts.RetryInterval())
-	assert.Equal(t, uint(3), opts.MaxRetries())
-	assert.Equal(t, uint(300000), opts.MaxRetryInterval())
+	assert.EqualValues(t, 5_000, opts.BatchSize())
+	assert.EqualValues(t, false, opts.UseGZip())
+	assert.EqualValues(t, 1_000, opts.FlushInterval())
+	assert.EqualValues(t, time.Nanosecond, opts.Precision())
+	assert.EqualValues(t, 50_000, opts.RetryBufferLimit())
+	assert.EqualValues(t, 5_000, opts.RetryInterval())
+	assert.EqualValues(t, 5, opts.MaxRetries())
+	assert.EqualValues(t, 125_000, opts.MaxRetryInterval())
+	assert.EqualValues(t, 180_000, opts.MaxRetryTime())
+	assert.EqualValues(t, 2, opts.ExponentialBase())
 	assert.Len(t, opts.DefaultTags(), 0)
 }
 
@@ -29,21 +31,25 @@ func TestSettingsOptions(t *testing.T) {
 	opts := write.DefaultOptions().
 		SetBatchSize(5).
 		SetUseGZip(true).
-		SetFlushInterval(5000).
+		SetFlushInterval(5_000).
 		SetPrecision(time.Millisecond).
 		SetRetryBufferLimit(5).
-		SetRetryInterval(1000).
+		SetRetryInterval(1_000).
 		SetMaxRetries(7).
-		SetMaxRetryInterval(150000).
+		SetMaxRetryInterval(150_000).
+		SetExponentialBase(3).
+		SetMaxRetryTime(200_000).
 		AddDefaultTag("a", "1").
 		AddDefaultTag("b", "2")
-	assert.Equal(t, uint(5), opts.BatchSize())
-	assert.Equal(t, true, opts.UseGZip())
-	assert.Equal(t, uint(5000), opts.FlushInterval())
-	assert.Equal(t, time.Millisecond, opts.Precision())
-	assert.Equal(t, uint(5), opts.RetryBufferLimit())
-	assert.Equal(t, uint(1000), opts.RetryInterval())
-	assert.Equal(t, uint(7), opts.MaxRetries())
-	assert.Equal(t, uint(150000), opts.MaxRetryInterval())
+	assert.EqualValues(t, 5, opts.BatchSize())
+	assert.EqualValues(t, true, opts.UseGZip())
+	assert.EqualValues(t, 5000, opts.FlushInterval())
+	assert.EqualValues(t, time.Millisecond, opts.Precision())
+	assert.EqualValues(t, 5, opts.RetryBufferLimit())
+	assert.EqualValues(t, 1000, opts.RetryInterval())
+	assert.EqualValues(t, 7, opts.MaxRetries())
+	assert.EqualValues(t, 150_000, opts.MaxRetryInterval())
+	assert.EqualValues(t, 200_000, opts.MaxRetryTime())
+	assert.EqualValues(t, 3, opts.ExponentialBase())
 	assert.Len(t, opts.DefaultTags(), 2)
 }
diff --git a/internal/write/service_test.go b/internal/write/service_test.go
index 08d0315..1cc7e3c 100644
--- a/internal/write/service_test.go
+++ b/internal/write/service_test.go
@@ -7,6 +7,7 @@ package write
 import (
 	"context"
 	"errors"
+	"fmt"
 	"strings"
 	"sync"
 	"testing"
@@ -20,6 +21,15 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+func TestPrecisionToString(t *testing.T) {
+	assert.Equal(t, "ns", precisionToString(time.Nanosecond))
+	assert.Equal(t, "us", precisionToString(time.Microsecond))
+	assert.Equal(t, "ms", precisionToString(time.Millisecond))
+	assert.Equal(t, "s", precisionToString(time.Second))
+	assert.Equal(t, "ns", precisionToString(time.Hour))
+	assert.Equal(t, "ns", precisionToString(time.Microsecond*20))
+}
+
 func TestAddDefaultTags(t *testing.T) {
 	hs := test.NewTestService(t, "http://localhost:8888")
 	opts := write.DefaultOptions()
@@ -67,77 +77,99 @@ func TestRetryStrategy(t *testing.T) {
 	opts := write.DefaultOptions().SetRetryInterval(1)
 	ctx := context.Background()
 	srv := NewService("my-org", "my-bucket", hs, opts)
+	// Set permanent reply error to force writes fail and retry
 	hs.SetReplyError(&http.Error{
 		StatusCode: 429,
 	})
-	b1 := NewBatch("1\n", opts.RetryInterval())
+	// This batch will fail and it be added to retry queue
+	b1 := NewBatch("1\n", opts.RetryInterval(), opts.MaxRetryTime())
 	err := srv.HandleWrite(ctx, b1)
 	assert.NotNil(t, err)
-	assert.Equal(t, uint(1), b1.retryDelay)
+	assert.EqualValues(t, 1, b1.retryDelay)
 	assert.Equal(t, 1, srv.retryQueue.list.Len())
 
 	//wait retry delay + little more
 	<-time.After(time.Millisecond*time.Duration(b1.retryDelay) + time.Microsecond*5)
-	b2 := NewBatch("2\n", opts.RetryInterval())
+	// First batch will be tried to write again and this one will added to retry queue
+	b2 := NewBatch("2\n", opts.RetryInterval(), opts.MaxRetryTime())
 	err = srv.HandleWrite(ctx, b2)
 	assert.NotNil(t, err)
-	assert.Equal(t, uint(5), b1.retryDelay)
+	assertBetween(t, b1.retryDelay, 2, 4)
 	assert.Equal(t, 2, srv.retryQueue.list.Len())
 
 	//wait retry delay + little more
 	<-time.After(time.Millisecond*time.Duration(b1.retryDelay) + time.Microsecond*5)
-	b3 := NewBatch("3\n", opts.RetryInterval())
+	// First batch will be tried to write again and this one will added to retry queue
+	b3 := NewBatch("3\n", opts.RetryInterval(), opts.MaxRetryTime())
 	err = srv.HandleWrite(ctx, b3)
 	assert.NotNil(t, err)
-	assert.Equal(t, uint(25), b1.retryDelay)
+	assertBetween(t, b1.retryDelay, 4, 8)
 	assert.Equal(t, 3, srv.retryQueue.list.Len())
 
-	// let write pass and it will clear queue
+	//wait retry delay + little more
+	<-time.After(time.Millisecond*time.Duration(b1.retryDelay) + time.Microsecond*5)
+	// First batch will be tried to write again and this one will added to retry queue
+	b4 := NewBatch("4\n", opts.RetryInterval(), opts.MaxRetryTime())
+	err = srv.HandleWrite(ctx, b4)
+	assert.NotNil(t, err)
+	assertBetween(t, b1.retryDelay, 8, 16)
+	assert.Equal(t, 4, srv.retryQueue.list.Len())
+
 	<-time.After(time.Millisecond*time.Duration(b1.retryDelay) + time.Microsecond*5)
+	// Clear error and let write pass
 	hs.SetReplyError(nil)
-	err = srv.HandleWrite(ctx, NewBatch("4\n", opts.RetryInterval()))
+	// Batches from retry queue will be sent first
+	err = srv.HandleWrite(ctx, NewBatch("5\n", opts.RetryInterval(), opts.MaxRetryTime()))
 	assert.Nil(t, err)
 	assert.Equal(t, 0, srv.retryQueue.list.Len())
-	require.Len(t, hs.Lines(), 4)
+	require.Len(t, hs.Lines(), 5)
 	assert.Equal(t, "1", hs.Lines()[0])
 	assert.Equal(t, "2", hs.Lines()[1])
 	assert.Equal(t, "3", hs.Lines()[2])
 	assert.Equal(t, "4", hs.Lines()[3])
+	assert.Equal(t, "5", hs.Lines()[4])
 }
 
 func TestBufferOverwrite(t *testing.T) {
 	log.Log.SetLogLevel(log.DebugLevel)
 	hs := test.NewTestService(t, "http://localhost:8086")
-	// Buffer limit 15000, bach ii 5000 => buffer for 3 batches
+	// Buffer limit 15000, bach is 5000 => buffer for 3 batches
 	opts := write.DefaultOptions().SetRetryInterval(1).SetRetryBufferLimit(15000)
 	ctx := context.Background()
 	srv := NewService("my-org", "my-bucket", hs, opts)
+	// Set permanent reply error to force writes fail and retry
 	hs.SetReplyError(&http.Error{
 		StatusCode: 429,
 	})
-	b1 := NewBatch("1\n", opts.RetryInterval())
+	// This batch will fail and it be added to retry queue
+	b1 := NewBatch("1\n", opts.RetryInterval(), opts.MaxRetryTime())
 	err := srv.HandleWrite(ctx, b1)
 	assert.NotNil(t, err)
 	assert.Equal(t, uint(1), b1.retryDelay)
 	assert.Equal(t, 1, srv.retryQueue.list.Len())
 
 	<-time.After(time.Millisecond * time.Duration(b1.retryDelay))
-	b2 := NewBatch("2\n", opts.RetryInterval())
+	b2 := NewBatch("2\n", opts.RetryInterval(), opts.MaxRetryTime())
+	// First batch will be tried to write again and this one will added to retry queue
 	err = srv.HandleWrite(ctx, b2)
 	assert.NotNil(t, err)
-	assert.Equal(t, uint(5), b1.retryDelay)
+	assertBetween(t, b1.retryDelay, 2, 4)
 	assert.Equal(t, 2, srv.retryQueue.list.Len())
 
 	<-time.After(time.Millisecond * time.Duration(b1.retryDelay))
-	b3 := NewBatch("3\n", opts.RetryInterval())
+	b3 := NewBatch("3\n", opts.RetryInterval(), opts.MaxRetryTime())
+	// First batch will be tried to write again and this one will added to retry queue
 	err = srv.HandleWrite(ctx, b3)
 	assert.NotNil(t, err)
-	assert.Equal(t, uint(25), b1.retryDelay)
+	assertBetween(t, b1.retryDelay, 4, 8)
 	assert.Equal(t, 3, srv.retryQueue.list.Len())
 
-	// Write early
-	<-time.After(time.Millisecond * time.Duration(b1.retryDelay) / 2)
-	b4 := NewBatch("4\n", opts.RetryInterval())
+	// Write early and overwrite
+	<-time.After(time.Millisecond)
+	b4 := NewBatch("4\n", opts.RetryInterval(), opts.MaxRetryTime())
+	// No write will occur, because retry delay has not passed yet
+	// However new bach will be added to retry queue. Retry queue has limit 3,
+	// so first batch will be discarded
 	err = srv.HandleWrite(ctx, b4)
 	assert.NoError(t, err)
 	assert.Equal(t, uint(1), b2.retryDelay)
@@ -145,16 +177,20 @@ func TestBufferOverwrite(t *testing.T) {
 
 	// Overwrite
 	<-time.After(time.Millisecond * time.Duration(b1.retryDelay) / 2)
-	b5 := NewBatch("5\n", opts.RetryInterval())
+	b5 := NewBatch("5\n", opts.RetryInterval(), opts.MaxRetryTime())
+	// Second batch will be tried to write again
+	// However, write will fail and as new batch is added to retry queue
+	// the second batch will be discarded
 	err = srv.HandleWrite(ctx, b5)
 	assert.Error(t, err)
-	assert.Equal(t, uint(5), b2.retryDelay)
+	assertBetween(t, b2.retryDelay, 2, 4)
 	assert.Equal(t, 3, srv.retryQueue.list.Len())
 
-	// let write pass and it will clear queue
 	<-time.After(time.Millisecond * time.Duration(b1.retryDelay))
+	// Clear error and let write pass
 	hs.SetReplyError(nil)
-	err = srv.HandleWrite(ctx, NewBatch("6\n", opts.RetryInterval()))
+	// Batches from retry queue will be sent first
+	err = srv.HandleWrite(ctx, NewBatch("6\n", opts.RetryInterval(), opts.MaxRetryTime()))
 	assert.Nil(t, err)
 	assert.Equal(t, 0, srv.retryQueue.list.Len())
 	require.Len(t, hs.Lines(), 4)
@@ -167,32 +203,130 @@ func TestBufferOverwrite(t *testing.T) {
 func TestMaxRetryInterval(t *testing.T) {
 	log.Log.SetLogLevel(log.DebugLevel)
 	hs := test.NewTestService(t, "http://localhost:8086")
-	//
-	opts := write.DefaultOptions().SetRetryInterval(1).SetMaxRetryInterval(10)
+	// MaxRetryInterval only 4ms, will be reached quickly
+	opts := write.DefaultOptions().SetRetryInterval(1).SetMaxRetryInterval(4)
 	ctx := context.Background()
 	srv := NewService("my-org", "my-bucket", hs, opts)
+	// Set permanent reply error to force writes fail and retry
 	hs.SetReplyError(&http.Error{
 		StatusCode: 503,
 	})
-	b1 := NewBatch("1\n", opts.RetryInterval())
+	// This batch will fail and it be added to retry queue
+	b1 := NewBatch("1\n", opts.RetryInterval(), opts.MaxRetryTime())
 	err := srv.HandleWrite(ctx, b1)
 	assert.NotNil(t, err)
 	assert.Equal(t, uint(1), b1.retryDelay)
 	assert.Equal(t, 1, srv.retryQueue.list.Len())
 
 	<-time.After(time.Millisecond * time.Duration(b1.retryDelay))
-	b2 := NewBatch("2\n", opts.RetryInterval())
+	b2 := NewBatch("2\n", opts.RetryInterval(), opts.MaxRetryTime())
+	// First batch will be tried to write again and this one will added to retry queue
 	err = srv.HandleWrite(ctx, b2)
 	assert.NotNil(t, err)
-	assert.Equal(t, uint(5), b1.retryDelay)
+	assertBetween(t, b1.retryDelay, 2, 4)
 	assert.Equal(t, 2, srv.retryQueue.list.Len())
 
 	<-time.After(time.Millisecond * time.Duration(b1.retryDelay))
-	b3 := NewBatch("3\n", opts.RetryInterval())
+	b3 := NewBatch("3\n", opts.RetryInterval(), opts.MaxRetryTime())
+	// First batch will be tried to write again and this one will added to retry queue
 	err = srv.HandleWrite(ctx, b3)
 	assert.NotNil(t, err)
-	assert.Equal(t, uint(10), b1.retryDelay)
+	// New computed delay of first batch should be 4-8, is limited to 4
+	assert.EqualValues(t, 4, b1.retryDelay)
 	assert.Equal(t, 3, srv.retryQueue.list.Len())
+
+	<-time.After(time.Millisecond * time.Duration(b1.retryDelay))
+	b4 := NewBatch("4\n", opts.RetryInterval(), opts.MaxRetryTime())
+	// First batch will be tried to write again and this one will added to retry queue
+	err = srv.HandleWrite(ctx, b4)
+	assert.NotNil(t, err)
+	// New computed delay of first batch should be 8-116, is limited to 4
+	assert.EqualValues(t, 4, b1.retryDelay)
+	assert.Equal(t, 4, srv.retryQueue.list.Len())
+}
+
+func TestMaxRetries(t *testing.T) {
+	log.Log.SetLogLevel(log.DebugLevel)
+	hs := test.NewTestService(t, "http://localhost:8086")
+	opts := write.DefaultOptions().SetRetryInterval(1)
+	ctx := context.Background()
+	srv := NewService("my-org", "my-bucket", hs, opts)
+	// Set permanent reply error to force writes fail and retry
+	hs.SetReplyError(&http.Error{
+		StatusCode: 429,
+	})
+	// This batch will fail and it be added to retry queue
+	b1 := NewBatch("1\n", opts.RetryInterval(), opts.MaxRetryTime())
+	err := srv.HandleWrite(ctx, b1)
+	assert.NotNil(t, err)
+	assert.EqualValues(t, 1, b1.retryDelay)
+	assert.Equal(t, 1, srv.retryQueue.list.Len())
+	// Write so many batches as it is maxRetries (5)
+	// First batch will be written and it will reach max retry limit
+	for i, e := uint(1), uint(2); i <= opts.MaxRetries(); i++ {
+		//wait retry delay + little more
+		<-time.After(time.Millisecond*time.Duration(b1.retryDelay) + time.Microsecond*5)
+		b := NewBatch(fmt.Sprintf("%d\n", i+1), opts.RetryInterval(), opts.MaxRetryTime())
+		err = srv.HandleWrite(ctx, b)
+		assert.NotNil(t, err)
+		assertBetween(t, b1.retryDelay, e, e*2)
+		exp := min(i+1, opts.MaxRetries())
+		assert.EqualValues(t, exp, srv.retryQueue.list.Len())
+		e *= 2
+	}
+	//Test if was removed from retry queue
+	assert.True(t, b1.evicted)
+
+	<-time.After(time.Millisecond*time.Duration(b1.retryDelay) + time.Microsecond*5)
+	// Clear error and let write pass
+	hs.SetReplyError(nil)
+	// Batches from retry queue will be sent first
+	err = srv.HandleWrite(ctx, NewBatch(fmt.Sprintf("%d\n", opts.MaxRetries()+2), opts.RetryInterval(), opts.MaxRetryTime()))
+	assert.Nil(t, err)
+	assert.Equal(t, 0, srv.retryQueue.list.Len())
+	require.Len(t, hs.Lines(), int(opts.MaxRetries()+1))
+	for i := uint(2); i <= opts.MaxRetries()+2; i++ {
+		assert.Equal(t, fmt.Sprintf("%d", i), hs.Lines()[i-2])
+	}
+}
+
+func TestMaxRetryTime(t *testing.T) {
+	log.Log.SetLogLevel(log.DebugLevel)
+	hs := test.NewTestService(t, "http://localhost:8086")
+	// Set maxRetryTime 5ms
+	opts := write.DefaultOptions().SetRetryInterval(1).SetMaxRetryTime(5)
+	ctx := context.Background()
+	srv := NewService("my-org", "my-bucket", hs, opts)
+	// Set permanent reply error to force writes fail and retry
+	hs.SetReplyError(&http.Error{
+		StatusCode: 429,
+	})
+	// This batch will fail and it be added to retry queue and it will expire 5ms after
+	b1 := NewBatch("1\n", opts.RetryInterval(), opts.MaxRetryTime())
+	err := srv.HandleWrite(ctx, b1)
+	assert.NotNil(t, err)
+	assert.EqualValues(t, 1, b1.retryDelay)
+	assert.Equal(t, 1, srv.retryQueue.list.Len())
+
+	// Wait for batch expiration
+	<-time.After(5 * time.Millisecond)
+	b := NewBatch("2\n", opts.RetryInterval(), opts.MaxRetryTime())
+	// First batch will be tried to write again and it will be checked agains maxRetryTime. New batch will added to retry queue
+	err = srv.HandleWrite(ctx, b)
+	require.NotNil(t, err)
+	// Error about batch expiration
+	assert.Equal(t, "write failed (attempts 1): max retry time exceeded", err.Error())
+	assert.Equal(t, 1, srv.retryQueue.list.Len())
+
+	// Clear error and let write pass
+	hs.SetReplyError(nil)
+	// A batch from retry queue will be sent first
+	err = srv.HandleWrite(ctx, NewBatch("3\n", opts.RetryInterval(), opts.MaxRetryTime()))
+	assert.Nil(t, err)
+	assert.Equal(t, 0, srv.retryQueue.list.Len())
+	require.Len(t, hs.Lines(), 2)
+	assert.Equal(t, "2", hs.Lines()[0])
+	assert.Equal(t, "3", hs.Lines()[1])
 }
 
 func TestRetryOnConnectionError(t *testing.T) {
@@ -203,34 +337,41 @@ func TestRetryOnConnectionError(t *testing.T) {
 	ctx := context.Background()
 	srv := NewService("my-org", "my-bucket", hs, opts)
 
+	// Set permanent non HTTP  error to force writes fail and retry
 	hs.SetReplyError(&http.Error{
 		Err: errors.New("connection refused"),
 	})
 
-	b1 := NewBatch("1\n", opts.RetryInterval())
+	// This batch will fail and it be added to retry queue
+	b1 := NewBatch("1\n", opts.RetryInterval(), opts.MaxRetryTime())
 	err := srv.HandleWrite(ctx, b1)
 	assert.NotNil(t, err)
-	assert.Equal(t, uint(1), b1.retryDelay)
+	assert.EqualValues(t, 1, b1.retryDelay)
 	assert.Equal(t, 1, srv.retryQueue.list.Len())
 
 	<-time.After(time.Millisecond * time.Duration(b1.retryDelay))
-	b2 := NewBatch("2\n", opts.RetryInterval())
+
+	b2 := NewBatch("2\n", opts.RetryInterval(), opts.MaxRetryTime())
+	// First batch will be tried to write again and this one will added to retry queue
 	err = srv.HandleWrite(ctx, b2)
 	assert.NotNil(t, err)
-	assert.Equal(t, uint(5), b1.retryDelay)
+	assertBetween(t, b1.retryDelay, 2, 4)
 	assert.Equal(t, 2, srv.retryQueue.list.Len())
 
 	<-time.After(time.Millisecond * time.Duration(b1.retryDelay))
-	b3 := NewBatch("3\n", opts.RetryInterval())
+
+	b3 := NewBatch("3\n", opts.RetryInterval(), opts.MaxRetryTime())
+	// First batch will be tried to write again and this one will added to retry queue
 	err = srv.HandleWrite(ctx, b3)
 	assert.NotNil(t, err)
-	assert.Equal(t, uint(25), b1.retryDelay)
+	assertBetween(t, b1.retryDelay, 4, 8)
 	assert.Equal(t, 3, srv.retryQueue.list.Len())
 
-	// let write pass and it will clear queue
 	<-time.After(time.Millisecond * time.Duration(b1.retryDelay))
+	// Clear error and let write pass
 	hs.SetReplyError(nil)
-	err = srv.HandleWrite(ctx, NewBatch("4\n", opts.RetryInterval()))
+	// Batches from retry queue will be sent first
+	err = srv.HandleWrite(ctx, NewBatch("4\n", opts.RetryInterval(), opts.MaxRetryTime()))
 	assert.Nil(t, err)
 	assert.Equal(t, 0, srv.retryQueue.list.Len())
 	require.Len(t, hs.Lines(), 4)
@@ -238,7 +379,6 @@ func TestRetryOnConnectionError(t *testing.T) {
 	assert.Equal(t, "2", hs.Lines()[1])
 	assert.Equal(t, "3", hs.Lines()[2])
 	assert.Equal(t, "4", hs.Lines()[3])
-
 }
 
 func TestNoRetryIfMaxRetriesIsZero(t *testing.T) {
@@ -253,7 +393,7 @@ func TestNoRetryIfMaxRetriesIsZero(t *testing.T) {
 		Err: errors.New("connection refused"),
 	})
 
-	b1 := NewBatch("1\n", opts.RetryInterval())
+	b1 := NewBatch("1\n", opts.RetryInterval(), opts.MaxRetryTime())
 	err := srv.HandleWrite(ctx, b1)
 	assert.NotNil(t, err)
 	assert.Equal(t, 0, srv.retryQueue.list.Len())
@@ -270,7 +410,7 @@ func TestWriteContextCancel(t *testing.T) {
 	wg.Add(1)
 	go func() {
 		<-time.After(10 * time.Millisecond)
-		err = srv.HandleWrite(ctx, NewBatch(strings.Join(lines, "\n"), opts.RetryInterval()))
+		err = srv.HandleWrite(ctx, NewBatch(strings.Join(lines, "\n"), opts.RetryInterval(), opts.MaxRetryTime()))
 		wg.Done()
 	}()
 	cancel()
@@ -278,3 +418,28 @@ func TestWriteContextCancel(t *testing.T) {
 	require.Equal(t, context.Canceled, err)
 	assert.Len(t, hs.Lines(), 0)
 }
+
+func TestPow(t *testing.T) {
+	assert.EqualValues(t, 1, pow(10, 0))
+	assert.EqualValues(t, 10, pow(10, 1))
+	assert.EqualValues(t, 4, pow(2, 2))
+	assert.EqualValues(t, 1, pow(1, 2))
+	assert.EqualValues(t, 125, pow(5, 3))
+}
+
+func assertBetween(t *testing.T, val, min, max uint) {
+	t.Helper()
+	assert.True(t, val >= min && val <= max, fmt.Sprintf("%d is outside <%d;%d>", val, min, max))
+}
+
+func TestComputeRetryDelay(t *testing.T) {
+	hs := test.NewTestService(t, "http://localhost:8888")
+	opts := write.DefaultOptions()
+	srv := NewService("my-org", "my-bucket", hs, opts)
+	assertBetween(t, srv.computeRetryDelay(0), 5_000, 10_000)
+	assertBetween(t, srv.computeRetryDelay(1), 10_000, 20_000)
+	assertBetween(t, srv.computeRetryDelay(2), 20_000, 40_000)
+	assertBetween(t, srv.computeRetryDelay(3), 40_000, 80_000)
+	assertBetween(t, srv.computeRetryDelay(4), 80_000, 125_000)
+	assert.EqualValues(t, 125_000, srv.computeRetryDelay(5))
+}
diff --git a/options_test.go b/options_test.go
index 82ccf3f..6afe193 100644
--- a/options_test.go
+++ b/options_test.go
@@ -20,17 +20,19 @@ import (
 
 func TestDefaultOptions(t *testing.T) {
 	opts := influxdb2.DefaultOptions()
-	assert.Equal(t, uint(5000), opts.BatchSize())
-	assert.Equal(t, false, opts.UseGZip())
-	assert.Equal(t, uint(1000), opts.FlushInterval())
-	assert.Equal(t, time.Nanosecond, opts.Precision())
-	assert.Equal(t, uint(50000), opts.RetryBufferLimit())
-	assert.Equal(t, uint(5000), opts.RetryInterval())
-	assert.Equal(t, uint(3), opts.MaxRetries())
-	assert.Equal(t, uint(300000), opts.MaxRetryInterval())
-	assert.Equal(t, (*tls.Config)(nil), opts.TLSConfig())
-	assert.Equal(t, uint(20), opts.HTTPRequestTimeout())
-	assert.Equal(t, uint(0), opts.LogLevel())
+	assert.EqualValues(t, 5_000, opts.BatchSize())
+	assert.EqualValues(t, false, opts.UseGZip())
+	assert.EqualValues(t, 1_000, opts.FlushInterval())
+	assert.EqualValues(t, time.Nanosecond, opts.Precision())
+	assert.EqualValues(t, 50_000, opts.RetryBufferLimit())
+	assert.EqualValues(t, 5_000, opts.RetryInterval())
+	assert.EqualValues(t, 5, opts.MaxRetries())
+	assert.EqualValues(t, 125_000, opts.MaxRetryInterval())
+	assert.EqualValues(t, 180_000, opts.MaxRetryTime())
+	assert.EqualValues(t, 2, opts.ExponentialBase())
+	assert.EqualValues(t, (*tls.Config)(nil), opts.TLSConfig())
+	assert.EqualValues(t, 20, opts.HTTPRequestTimeout())
+	assert.EqualValues(t, 0, opts.LogLevel())
 }
 
 func TestSettingsOptions(t *testing.T) {
@@ -40,31 +42,35 @@ func TestSettingsOptions(t *testing.T) {
 	opts := influxdb2.DefaultOptions().
 		SetBatchSize(5).
 		SetUseGZip(true).
-		SetFlushInterval(5000).
+		SetFlushInterval(5_000).
 		SetPrecision(time.Millisecond).
 		SetRetryBufferLimit(5).
-		SetRetryInterval(1000).
-		SetMaxRetryInterval(10000).
+		SetRetryInterval(1_000).
+		SetMaxRetryInterval(10_000).
 		SetMaxRetries(7).
+		SetMaxRetryTime(500_000).
+		SetExponentialBase(5).
 		SetTLSConfig(tlsConfig).
 		SetHTTPRequestTimeout(50).
 		SetLogLevel(3).
 		AddDefaultTag("t", "a")
-	assert.Equal(t, uint(5), opts.BatchSize())
-	assert.Equal(t, true, opts.UseGZip())
-	assert.Equal(t, uint(5000), opts.FlushInterval())
-	assert.Equal(t, time.Millisecond, opts.Precision())
-	assert.Equal(t, uint(5), opts.RetryBufferLimit())
-	assert.Equal(t, uint(1000), opts.RetryInterval())
-	assert.Equal(t, uint(10000), opts.MaxRetryInterval())
-	assert.Equal(t, uint(7), opts.MaxRetries())
-	assert.Equal(t, tlsConfig, opts.TLSConfig())
-	assert.Equal(t, uint(50), opts.HTTPRequestTimeout())
+	assert.EqualValues(t, 5, opts.BatchSize())
+	assert.EqualValues(t, true, opts.UseGZip())
+	assert.EqualValues(t, 5_000, opts.FlushInterval())
+	assert.EqualValues(t, time.Millisecond, opts.Precision())
+	assert.EqualValues(t, 5, opts.RetryBufferLimit())
+	assert.EqualValues(t, 1_000, opts.RetryInterval())
+	assert.EqualValues(t, 10_000, opts.MaxRetryInterval())
+	assert.EqualValues(t, 7, opts.MaxRetries())
+	assert.EqualValues(t, 500_000, opts.MaxRetryTime())
+	assert.EqualValues(t, 5, opts.ExponentialBase())
+	assert.EqualValues(t, tlsConfig, opts.TLSConfig())
+	assert.EqualValues(t, 50, opts.HTTPRequestTimeout())
 	if client := opts.HTTPClient(); assert.NotNil(t, client) {
-		assert.Equal(t, 50*time.Second, client.Timeout)
+		assert.EqualValues(t, 50*time.Second, client.Timeout)
 		assert.Equal(t, tlsConfig, client.Transport.(*http.Transport).TLSClientConfig)
 	}
-	assert.Equal(t, uint(3), opts.LogLevel())
+	assert.EqualValues(t, 3, opts.LogLevel())
 	assert.Len(t, opts.WriteOptions().DefaultTags(), 1)
 
 	client := &http.Client{