Repository: influxdata/influxdb-client-go
Pull request #179 (2020-08-13 08:48:37)

feat: Unifying retry strategy among InfluxDB 2 clients
## Proposed Changes

- Added exponential backoff
- Added MaxRetryInterval to set maximum retry delay
- Various retry logic fixes
- Added more tests

## Checklist

- [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 4621616..8c84874 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,6 +6,7 @@
 
 ### Features
 1. [#165](https://github.com/influxdata/influxdb-client-go/pull/165) Allow overriding the http.Client for the http service.
+1. [#179](https://github.com/influxdata/influxdb-client-go/pull/179) Unifying retry strategy among InfluxDB 2 clients: added exponential backoff. 
 
 ### Bug fixes 
 1. [#175](https://github.com/influxdata/influxdb-client-go/pull/175) Fixed WriteAPIs management. Keeping single instance for each org and bucket pair.
diff --git a/api/write/options.go b/api/write/options.go
index db14cd5..80b5cea 100644
--- a/api/write/options.go
+++ b/api/write/options.go
@@ -14,12 +14,6 @@ type Options struct {
 	batchSize uint
 	// Interval, in ms, in which is buffer flushed if it has not been already written (by reaching batch size) . Default 1000ms
 	flushInterval uint
-	// Default retry interval in ms, if not sent by server. Default 1000ms
-	retryInterval uint
-	// Maximum count of retry attempts of failed writes
-	maxRetries uint
-	// Maximum number of points to keep for retry. Should be multiple of BatchSize. Default 10,000
-	retryBufferLimit uint
 	// Precision to use in writes for timestamp. In unit of duration: time.Nanosecond, time.Microsecond, time.Millisecond, time.Second
 	// Default time.Nanosecond
 	precision time.Duration
@@ -27,6 +21,14 @@ 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
+	retryInterval uint
+	// Maximum count of retry attempts of failed writes
+	maxRetries uint
+	// 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)
+	maxRetryInterval uint
 }
 
 // BatchSize returns size of batch
@@ -84,6 +86,17 @@ func (o *Options) SetRetryBufferLimit(retryBufferLimit uint) *Options {
 	return o
 }
 
+// MaxRetryInterval return maximum retry interval in ms. Default 5min.
+func (o *Options) MaxRetryInterval() uint {
+	return o.maxRetryInterval
+}
+
+// SetMaxRetryInterval set maximum retry interval in ms
+func (o *Options) SetMaxRetryInterval(maxRetryIntervalMs uint) *Options {
+	o.maxRetryInterval = maxRetryIntervalMs
+	return o
+}
+
 // Precision returns time precision for writes
 func (o *Options) Precision() time.Duration {
 	return o.precision
@@ -124,5 +137,5 @@ func (o *Options) DefaultTags() map[string]string {
 
 // DefaultOptions returns Options object with default values
 func DefaultOptions() *Options {
-	return &Options{batchSize: 5000, maxRetries: 3, retryInterval: 1000, flushInterval: 1000, precision: time.Nanosecond, useGZip: false, retryBufferLimit: 10000, defaultTags: make(map[string]string)}
+	return &Options{batchSize: 5000, maxRetries: 3, retryInterval: 5000, maxRetryInterval: 300000, flushInterval: 1000, precision: time.Nanosecond, useGZip: false, retryBufferLimit: 50000, defaultTags: make(map[string]string)}
 }
diff --git a/internal/write/queue.go b/internal/write/queue.go
index f467ac8..11d3345 100644
--- a/internal/write/queue.go
+++ b/internal/write/queue.go
@@ -30,7 +30,9 @@ func (q *queue) pop() *Batch {
 	el := q.list.Front()
 	if el != nil {
 		q.list.Remove(el)
-		return el.Value.(*Batch)
+		batch := el.Value.(*Batch)
+		batch.evicted = true
+		return batch
 	}
 	return nil
 }
diff --git a/internal/write/service.go b/internal/write/service.go
index d1d5b46..238a3b1 100644
--- a/internal/write/service.go
+++ b/internal/write/service.go
@@ -25,26 +25,28 @@ import (
 
 type Batch struct {
 	batch         string
-	retryInterval uint
-	retries       uint
+	retryDelay    uint
+	retryAttempts uint
+	evicted       bool
 }
 
-func NewBatch(data string, retryInterval uint) *Batch {
+func NewBatch(data string, retryDelay uint) *Batch {
 	return &Batch{
-		batch:         data,
-		retryInterval: retryInterval,
+		batch:      data,
+		retryDelay: retryDelay,
 	}
 }
 
 type Service struct {
-	org              string
-	bucket           string
-	httpService      ihttp.Service
-	url              string
-	lastWriteAttempt time.Time
-	retryQueue       *queue
-	lock             sync.Mutex
-	writeOptions     *write.Options
+	org                  string
+	bucket               string
+	httpService          ihttp.Service
+	url                  string
+	lastWriteAttempt     time.Time
+	retryQueue           *queue
+	lock                 sync.Mutex
+	writeOptions         *write.Options
+	retryExponentialBase uint
 }
 
 func NewService(org string, bucket string, httpService ihttp.Service, options *write.Options) *Service {
@@ -53,7 +55,7 @@ func NewService(org string, bucket string, httpService ihttp.Service, options *w
 	if retryBufferLimit == 0 {
 		retryBufferLimit = 1
 	}
-	return &Service{org: org, bucket: bucket, httpService: httpService, writeOptions: options, retryQueue: newQueue(int(retryBufferLimit))}
+	return &Service{org: org, bucket: bucket, httpService: httpService, writeOptions: options, retryQueue: newQueue(int(retryBufferLimit)), retryExponentialBase: 5}
 }
 
 func (w *Service) HandleWrite(ctx context.Context, batch *Batch) error {
@@ -72,18 +74,20 @@ func (w *Service) HandleWrite(ctx context.Context, batch *Batch) error {
 			if !retrying {
 				b := w.retryQueue.first()
 				// Can we write? In case of retryable error we must wait a bit
-				if w.lastWriteAttempt.IsZero() || time.Now().After(w.lastWriteAttempt.Add(time.Millisecond*time.Duration(b.retryInterval))) {
+				if w.lastWriteAttempt.IsZero() || time.Now().After(w.lastWriteAttempt.Add(time.Millisecond*time.Duration(b.retryDelay))) {
 					retrying = true
 				} else {
 					log.Log.Warn("Write proc: cannot write yet, storing batch to queue")
-					w.retryQueue.push(batch)
+					if w.retryQueue.push(batch) {
+						log.Log.Warn("Write proc: Retry buffer full, discarding oldest batch")
+					}
 					batchToWrite = nil
 				}
 			}
 			if retrying {
-				batchToWrite = w.retryQueue.pop()
-				batchToWrite.retries++
-				if batch != nil {
+				batchToWrite = w.retryQueue.first()
+				batchToWrite.retryAttempts++
+				if batch != nil { //store actual batch to retry queue
 					if w.retryQueue.push(batch) {
 						log.Log.Warn("Write proc: Retry buffer full, discarding oldest batch")
 					}
@@ -91,11 +95,40 @@ func (w *Service) HandleWrite(ctx context.Context, batch *Batch) error {
 				}
 			}
 		}
+		// write batch
 		if batchToWrite != nil {
-			err := w.WriteBatch(ctx, batchToWrite)
-			batchToWrite = nil
-			if err != nil {
-				return err
+			perror := w.WriteBatch(ctx, batchToWrite)
+			if perror != nil {
+				if perror.StatusCode >= http.StatusTooManyRequests {
+					log.Log.Errorf("Write error: %s\nBatch kept for retrying\n", perror.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())
+					}
+					if batchToWrite.retryAttempts == 0 {
+						if w.retryQueue.push(batch) {
+							log.Log.Warn("Retry buffer full, discarding oldest batch")
+						}
+					} else if batchToWrite.retryAttempts == w.writeOptions.MaxRetries() {
+						log.Log.Warn("Reached maximum number of retries, discarding batch")
+						if !batchToWrite.evicted {
+							w.retryQueue.pop()
+						}
+					}
+				} else {
+					log.Log.Errorf("Write error: %s\n", perror.Error())
+				}
+				return perror
+			} else {
+				if retrying && !batchToWrite.evicted {
+					w.retryQueue.pop()
+				}
+				batchToWrite = nil
 			}
 		} else {
 			break
@@ -104,11 +137,11 @@ func (w *Service) HandleWrite(ctx context.Context, batch *Batch) error {
 	return nil
 }
 
-func (w *Service) WriteBatch(ctx context.Context, batch *Batch) error {
+func (w *Service) WriteBatch(ctx context.Context, batch *Batch) *ihttp.Error {
 	wURL, err := w.WriteURL()
 	if err != nil {
 		log.Log.Errorf("%s\n", err.Error())
-		return err
+		return ihttp.NewError(err)
 	}
 	var body io.Reader
 	body = strings.NewReader(batch.batch)
@@ -116,7 +149,7 @@ func (w *Service) WriteBatch(ctx context.Context, batch *Batch) error {
 	if w.writeOptions.UseGZip() {
 		body, err = gzip.CompressWithGzip(body)
 		if err != nil {
-			return err
+			return ihttp.NewError(err)
 		}
 	}
 	w.lastWriteAttempt = time.Now()
@@ -126,30 +159,12 @@ func (w *Service) WriteBatch(ctx context.Context, batch *Batch) error {
 		}
 	}, func(r *http.Response) error {
 		// discard body so connection can be reused
-		//_, _ = io.Copy(ioutil.Discard, r.Body)
-		//_ = r.Body.Close()
+		// _, _ = io.Copy(ioutil.Discard, r.Body)
+		// _ = r.Body.Close()
 		return nil
 	})
 
-	if perror != nil {
-		if perror.StatusCode == http.StatusTooManyRequests || perror.StatusCode == http.StatusServiceUnavailable {
-			log.Log.Errorf("Write error: %s\nBatch kept for retrying\n", perror.Error())
-			if perror.RetryAfter > 0 {
-				batch.retryInterval = perror.RetryAfter * 1000
-			} else {
-				batch.retryInterval = w.writeOptions.RetryInterval()
-			}
-			if batch.retries < w.writeOptions.MaxRetries() {
-				if w.retryQueue.push(batch) {
-					log.Log.Warn("Retry buffer full, discarding oldest batch")
-				}
-			}
-		} else {
-			log.Log.Errorf("Write error: %s\n", perror.Error())
-		}
-		return perror
-	}
-	return nil
+	return perror
 }
 
 type pointWithDefaultTags struct {
@@ -258,3 +273,10 @@ func precisionToString(precision time.Duration) string {
 	}
 	return prec
 }
+
+func min(a, b uint) uint {
+	if a > b {
+		return b
+	}
+	return a
+}
diff --git a/options.go b/options.go
index 6f2feb0..9e4326a 100644
--- a/options.go
+++ b/options.go
@@ -78,6 +78,17 @@ func (o *Options) SetRetryBufferLimit(retryBufferLimit uint) *Options {
 	return o
 }
 
+// MaxRetryInterval return maximum retry interval in ms. Default 5min.
+func (o *Options) MaxRetryInterval() uint {
+	return o.WriteOptions().MaxRetryInterval()
+}
+
+// SetMaxRetryInterval set maximum retry interval in ms
+func (o *Options) SetMaxRetryInterval(maxRetryIntervalMs uint) *Options {
+	o.WriteOptions().SetMaxRetryInterval(maxRetryIntervalMs)
+	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 aa08d23..5c2a1ee 100644
--- a/api/write/options_test.go
+++ b/api/write/options_test.go
@@ -18,9 +18,10 @@ func TestDefaultOptions(t *testing.T) {
 	assert.Equal(t, false, opts.UseGZip())
 	assert.Equal(t, uint(1000), opts.FlushInterval())
 	assert.Equal(t, time.Nanosecond, opts.Precision())
-	assert.Equal(t, uint(10000), opts.RetryBufferLimit())
-	assert.Equal(t, uint(1000), opts.RetryInterval())
+	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.Len(t, opts.DefaultTags(), 0)
 }
 
@@ -31,8 +32,9 @@ func TestSettingsOptions(t *testing.T) {
 		SetFlushInterval(5000).
 		SetPrecision(time.Millisecond).
 		SetRetryBufferLimit(5).
-		SetRetryInterval(5000).
+		SetRetryInterval(1000).
 		SetMaxRetries(7).
+		SetMaxRetryInterval(150000).
 		AddDefaultTag("a", "1").
 		AddDefaultTag("b", "2")
 	assert.Equal(t, uint(5), opts.BatchSize())
@@ -40,7 +42,8 @@ func TestSettingsOptions(t *testing.T) {
 	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(5000), opts.RetryInterval())
+	assert.Equal(t, uint(1000), opts.RetryInterval())
 	assert.Equal(t, uint(7), opts.MaxRetries())
+	assert.Equal(t, uint(150000), opts.MaxRetryInterval())
 	assert.Len(t, opts.DefaultTags(), 2)
 }
diff --git a/api/writeAPIBlocking_test.go b/api/writeAPIBlocking_test.go
index bff0c99..fd7d8e9 100644
--- a/api/writeAPIBlocking_test.go
+++ b/api/writeAPIBlocking_test.go
@@ -6,56 +6,56 @@ package api
 
 import (
 	"context"
-
 	"sync"
 	"testing"
 	"time"
 
 	"github.com/influxdata/influxdb-client-go/api/write"
 	"github.com/influxdata/influxdb-client-go/internal/http"
+	"github.com/influxdata/influxdb-client-go/internal/test"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
 func TestWritePoint(t *testing.T) {
-	service := newTestService(t, "http://localhost:8888")
+	service := test.NewTestService(t, "http://localhost:8888")
 	writeAPI := NewWriteAPIBlocking("my-org", "my-bucket", service, write.DefaultOptions().SetBatchSize(5))
 	points := genPoints(10)
 	err := writeAPI.WritePoint(context.Background(), points...)
 	require.Nil(t, err)
-	require.Len(t, service.lines, 10)
+	require.Len(t, service.Lines(), 10)
 	for i, p := range points {
 		line := write.PointToLineProtocol(p, writeAPI.writeOptions.Precision())
 		//cut off last \n char
 		line = line[:len(line)-1]
-		assert.Equal(t, service.lines[i], line)
+		assert.Equal(t, service.Lines()[i], line)
 	}
 }
 
 func TestWriteRecord(t *testing.T) {
-	service := newTestService(t, "http://localhost:8888")
+	service := test.NewTestService(t, "http://localhost:8888")
 	writeAPI := NewWriteAPIBlocking("my-org", "my-bucket", service, write.DefaultOptions().SetBatchSize(5))
 	lines := genRecords(10)
 	err := writeAPI.WriteRecord(context.Background(), lines...)
 	require.Nil(t, err)
-	require.Len(t, service.lines, 10)
+	require.Len(t, service.Lines(), 10)
 	for i, l := range lines {
-		assert.Equal(t, l, service.lines[i])
+		assert.Equal(t, l, service.Lines()[i])
 	}
 	service.Close()
 
 	err = writeAPI.WriteRecord(context.Background())
 	require.Nil(t, err)
-	require.Len(t, service.lines, 0)
+	require.Len(t, service.Lines(), 0)
 
-	service.replyError = &http.Error{Code: "invalid", Message: "data"}
+	service.SetReplyError(&http.Error{Code: "invalid", Message: "data"})
 	err = writeAPI.WriteRecord(context.Background(), lines...)
 	require.NotNil(t, err)
 	require.Equal(t, "invalid: data", err.Error())
 }
 
 func TestWriteContextCancel(t *testing.T) {
-	service := newTestService(t, "http://localhost:8888")
+	service := test.NewTestService(t, "http://localhost:8888")
 	writeAPI := NewWriteAPIBlocking("my-org", "my-bucket", service, write.DefaultOptions().SetBatchSize(5))
 	lines := genRecords(10)
 	ctx, cancel := context.WithCancel(context.Background())
@@ -70,5 +70,5 @@ func TestWriteContextCancel(t *testing.T) {
 	cancel()
 	wg.Wait()
 	require.Equal(t, context.Canceled, err)
-	assert.Len(t, service.lines, 0)
+	assert.Len(t, service.Lines(), 0)
 }
diff --git a/api/write_test.go b/api/write_test.go
index ac72521..21f581a 100644
--- a/api/write_test.go
+++ b/api/write_test.go
@@ -5,13 +5,8 @@
 package api
 
 import (
-	"compress/gzip"
-	"context"
 	"fmt"
-	"io"
-	"io/ioutil"
 	"math/rand"
-	"net/http"
 	"strings"
 	"sync"
 	"testing"
@@ -20,124 +15,11 @@ import (
 	"github.com/influxdata/influxdb-client-go/api/write"
 	ihttp "github.com/influxdata/influxdb-client-go/internal/http"
 	"github.com/influxdata/influxdb-client-go/internal/log"
+	"github.com/influxdata/influxdb-client-go/internal/test"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
-type testHTTPService struct {
-	serverURL      string
-	authorization  string
-	lines          []string
-	t              *testing.T
-	wasGzip        bool
-	requestHandler func(c *testHTTPService, url string, body io.Reader) error
-	replyError     *ihttp.Error
-	lock           sync.Mutex
-}
-
-func (t *testHTTPService) ServerURL() string {
-	return t.serverURL
-}
-
-func (t *testHTTPService) ServerAPIURL() string {
-	return t.serverURL
-}
-
-func (t *testHTTPService) Authorization() string {
-	return t.authorization
-}
-
-func (t *testHTTPService) HTTPClient() *http.Client {
-	return nil
-}
-
-func (t *testHTTPService) Close() {
-	t.lock.Lock()
-	if len(t.lines) > 0 {
-		t.lines = t.lines[:0]
-	}
-	t.wasGzip = false
-	t.replyError = nil
-	t.requestHandler = nil
-	t.lock.Unlock()
-}
-
-func (t *testHTTPService) ReplyError() *ihttp.Error {
-	t.lock.Lock()
-	defer t.lock.Unlock()
-	return t.replyError
-}
-
-func (t *testHTTPService) SetAuthorization(_ string) {
-
-}
-func (t *testHTTPService) GetRequest(_ context.Context, _ string, _ ihttp.RequestCallback, _ ihttp.ResponseCallback) *ihttp.Error {
-	return nil
-}
-func (t *testHTTPService) DoHTTPRequest(_ *http.Request, _ ihttp.RequestCallback, _ ihttp.ResponseCallback) *ihttp.Error {
-	return nil
-}
-
-func (t *testHTTPService) DoHTTPRequestWithResponse(_ *http.Request, _ ihttp.RequestCallback) (*http.Response, error) {
-	return nil, nil
-}
-
-func (t *testHTTPService) PostRequest(_ context.Context, url string, body io.Reader, requestCallback ihttp.RequestCallback, _ ihttp.ResponseCallback) *ihttp.Error {
-	req, err := http.NewRequest("POST", url, nil)
-	if err != nil {
-		return ihttp.NewError(err)
-	}
-	if requestCallback != nil {
-		requestCallback(req)
-	}
-	if req.Header.Get("Content-Encoding") == "gzip" {
-		body, _ = gzip.NewReader(body)
-		t.wasGzip = true
-	}
-	assert.Equal(t.t, fmt.Sprintf("%swrite?bucket=my-bucket&org=my-org&precision=ns", t.serverURL), url)
-
-	if t.ReplyError() != nil {
-		return t.ReplyError()
-	}
-	if t.requestHandler != nil {
-		err = t.requestHandler(t, url, body)
-	} else {
-		err = t.decodeLines(body)
-	}
-
-	if err != nil {
-		return ihttp.NewError(err)
-	} else {
-		return nil
-	}
-}
-
-func (t *testHTTPService) decodeLines(body io.Reader) error {
-	bytes, err := ioutil.ReadAll(body)
-	if err != nil {
-		return err
-	}
-	lines := strings.Split(string(bytes), "\n")
-	lines = lines[:len(lines)-1]
-	t.lock.Lock()
-	t.lines = append(t.lines, lines...)
-	t.lock.Unlock()
-	return nil
-}
-
-func (t *testHTTPService) Lines() []string {
-	t.lock.Lock()
-	defer t.lock.Unlock()
-	return t.lines
-}
-
-func newTestService(t *testing.T, serverURL string) *testHTTPService {
-	return &testHTTPService{
-		t:         t,
-		serverURL: serverURL + "/api/v2/",
-	}
-}
-
 func genPoints(num int) []*write.Point {
 	points := make([]*write.Point, num)
 	rand.Seed(321)
@@ -182,7 +64,7 @@ func genRecords(num int) []string {
 }
 
 func TestWriteAPIWriteDefaultTag(t *testing.T) {
-	service := newTestService(t, "http://localhost:8888")
+	service := test.NewTestService(t, "http://localhost:8888")
 	opts := write.DefaultOptions().
 		SetBatchSize(1)
 	opts.AddDefaultTag("dft", "a")
@@ -201,7 +83,7 @@ func TestWriteAPIWriteDefaultTag(t *testing.T) {
 }
 
 func TestWriteAPIImpl_Write(t *testing.T) {
-	service := newTestService(t, "http://localhost:8888")
+	service := test.NewTestService(t, "http://localhost:8888")
 	writeAPI := NewWriteAPI("my-org", "my-bucket", service, write.DefaultOptions().SetBatchSize(5))
 	points := genPoints(10)
 	for _, p := range points {
@@ -218,7 +100,7 @@ func TestWriteAPIImpl_Write(t *testing.T) {
 }
 
 func TestGzipWithFlushing(t *testing.T) {
-	service := newTestService(t, "http://localhost:8888")
+	service := test.NewTestService(t, "http://localhost:8888")
 	log.Log.SetDebugLevel(4)
 	writeAPI := NewWriteAPI("my-org", "my-bucket", service, write.DefaultOptions().SetBatchSize(5).SetUseGZip(true))
 	points := genPoints(5)
@@ -230,7 +112,7 @@ func TestGzipWithFlushing(t *testing.T) {
 	end := time.Now()
 	fmt.Printf("Flash duration: %dns\n", end.Sub(start).Nanoseconds())
 	assert.Len(t, service.Lines(), 5)
-	assert.True(t, service.wasGzip)
+	assert.True(t, service.WasGzip())
 
 	service.Close()
 	writeAPI.writeOptions.SetUseGZip(false)
@@ -239,12 +121,12 @@ func TestGzipWithFlushing(t *testing.T) {
 	}
 	writeAPI.waitForFlushing()
 	assert.Len(t, service.Lines(), 5)
-	assert.False(t, service.wasGzip)
+	assert.False(t, service.WasGzip())
 
 	writeAPI.Close()
 }
 func TestFlushInterval(t *testing.T) {
-	service := newTestService(t, "http://localhost:8888")
+	service := test.NewTestService(t, "http://localhost:8888")
 	writeAPI := NewWriteAPI("my-org", "my-bucket", service, write.DefaultOptions().SetBatchSize(10).SetFlushInterval(500))
 	points := genPoints(5)
 	for _, p := range points {
@@ -268,7 +150,7 @@ func TestFlushInterval(t *testing.T) {
 }
 
 func TestRetry(t *testing.T) {
-	service := newTestService(t, "http://localhost:8888")
+	service := test.NewTestService(t, "http://localhost:8888")
 	log.Log.SetDebugLevel(5)
 	writeAPI := NewWriteAPI("my-org", "my-bucket", service, write.DefaultOptions().SetBatchSize(5).SetRetryInterval(10000))
 	points := genPoints(15)
@@ -278,10 +160,10 @@ func TestRetry(t *testing.T) {
 	writeAPI.waitForFlushing()
 	require.Len(t, service.Lines(), 5)
 	service.Close()
-	service.replyError = &ihttp.Error{
+	service.SetReplyError(&ihttp.Error{
 		StatusCode: 429,
 		RetryAfter: 5,
-	}
+	})
 	for i := 0; i < 5; i++ {
 		writeAPI.WritePoint(points[i])
 	}
@@ -305,13 +187,13 @@ func TestRetry(t *testing.T) {
 }
 
 func TestWriteError(t *testing.T) {
-	service := newTestService(t, "http://localhost:8888")
+	service := test.NewTestService(t, "http://localhost:8888")
 	log.Log.SetDebugLevel(3)
-	service.replyError = &ihttp.Error{
+	service.SetReplyError(&ihttp.Error{
 		StatusCode: 400,
 		Code:       "write",
 		Message:    "error",
-	}
+	})
 	writeAPI := NewWriteAPI("my-org", "my-bucket", service, write.DefaultOptions().SetBatchSize(5))
 	errCh := writeAPI.Errors()
 	var recErr error
diff --git a/internal/test/http_service.go b/internal/test/http_service.go
new file mode 100644
index 0000000..c64a1c4
--- /dev/null
+++ b/internal/test/http_service.go
@@ -0,0 +1,148 @@
+// Copyright 2020 InfluxData, Inc. All rights reserved.
+// Use of this source code is governed by MIT
+// license that can be found in the LICENSE file.
+
+// Package test provides shared test utils
+package test
+
+import (
+	"compress/gzip"
+	"context"
+	"fmt"
+	ihttp "github.com/influxdata/influxdb-client-go/internal/http"
+	"github.com/stretchr/testify/assert"
+	"io"
+	"io/ioutil"
+	"net/http"
+	"strings"
+	"sync"
+	"testing"
+)
+
+type HTTPService struct {
+	serverURL      string
+	authorization  string
+	lines          []string
+	t              *testing.T
+	wasGzip        bool
+	requestHandler func(c *HTTPService, url string, body io.Reader) error
+	replyError     *ihttp.Error
+	lock           sync.Mutex
+}
+
+func (t *HTTPService) WasGzip() bool {
+	return t.wasGzip
+}
+
+func (t *HTTPService) SetWasGzip(wasGzip bool) {
+	t.wasGzip = wasGzip
+}
+
+func (t *HTTPService) ServerURL() string {
+	return t.serverURL
+}
+
+func (t *HTTPService) ServerAPIURL() string {
+	return t.serverURL
+}
+
+func (t *HTTPService) Authorization() string {
+	return t.authorization
+}
+
+func (t *HTTPService) HTTPClient() *http.Client {
+	return nil
+}
+
+func (t *HTTPService) Close() {
+	t.lock.Lock()
+	if len(t.lines) > 0 {
+		t.lines = t.lines[:0]
+	}
+	t.wasGzip = false
+	t.replyError = nil
+	t.requestHandler = nil
+	t.lock.Unlock()
+}
+
+func (t *HTTPService) SetReplyError(replyError *ihttp.Error) {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+	t.replyError = replyError
+}
+
+func (t *HTTPService) ReplyError() *ihttp.Error {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+	return t.replyError
+}
+
+func (t *HTTPService) SetAuthorization(_ string) {
+
+}
+func (t *HTTPService) GetRequest(_ context.Context, _ string, _ ihttp.RequestCallback, _ ihttp.ResponseCallback) *ihttp.Error {
+	return nil
+}
+func (t *HTTPService) DoHTTPRequest(_ *http.Request, _ ihttp.RequestCallback, _ ihttp.ResponseCallback) *ihttp.Error {
+	return nil
+}
+
+func (t *HTTPService) DoHTTPRequestWithResponse(_ *http.Request, _ ihttp.RequestCallback) (*http.Response, error) {
+	return nil, nil
+}
+
+func (t *HTTPService) PostRequest(_ context.Context, url string, body io.Reader, requestCallback ihttp.RequestCallback, _ ihttp.ResponseCallback) *ihttp.Error {
+	req, err := http.NewRequest("POST", url, nil)
+	if err != nil {
+		return ihttp.NewError(err)
+	}
+	if requestCallback != nil {
+		requestCallback(req)
+	}
+	if req.Header.Get("Content-Encoding") == "gzip" {
+		body, _ = gzip.NewReader(body)
+		t.wasGzip = true
+	}
+	assert.Equal(t.t, fmt.Sprintf("%swrite?bucket=my-bucket&org=my-org&precision=ns", t.serverURL), url)
+
+	if t.ReplyError() != nil {
+		return t.ReplyError()
+	}
+	if t.requestHandler != nil {
+		err = t.requestHandler(t, url, body)
+	} else {
+		err = t.decodeLines(body)
+	}
+
+	if err != nil {
+		return ihttp.NewError(err)
+	} else {
+		return nil
+	}
+}
+
+func (t *HTTPService) decodeLines(body io.Reader) error {
+	bytes, err := ioutil.ReadAll(body)
+	if err != nil {
+		return err
+	}
+	lines := strings.Split(string(bytes), "\n")
+	lines = lines[:len(lines)-1]
+	t.lock.Lock()
+	t.lines = append(t.lines, lines...)
+	t.lock.Unlock()
+	return nil
+}
+
+func (t *HTTPService) Lines() []string {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+	return t.lines
+}
+
+func NewTestService(t *testing.T, serverURL string) *HTTPService {
+	return &HTTPService{
+		t:         t,
+		serverURL: serverURL + "/api/v2/",
+	}
+}
diff --git a/internal/write/queue_test.go b/internal/write/queue_test.go
index 40175b2..9acf7a1 100644
--- a/internal/write/queue_test.go
+++ b/internal/write/queue_test.go
@@ -13,7 +13,7 @@ import (
 func TestQueue(t *testing.T) {
 	que := newQueue(2)
 	assert.True(t, que.isEmpty())
-	b := &Batch{batch: "batch", retryInterval: 3, retries: 3}
+	b := &Batch{batch: "batch", retryDelay: 3, retryAttempts: 3}
 	que.push(b)
 	assert.False(t, que.isEmpty())
 	b2 := que.pop()
diff --git a/internal/write/service_test.go b/internal/write/service_test.go
index 5783df6..6653785 100644
--- a/internal/write/service_test.go
+++ b/internal/write/service_test.go
@@ -5,10 +5,16 @@
 package write
 
 import (
+	"context"
+	"testing"
+	"time"
+
 	"github.com/influxdata/influxdb-client-go/api/write"
+	ihttp "github.com/influxdata/influxdb-client-go/internal/http"
+	"github.com/influxdata/influxdb-client-go/internal/log"
+	"github.com/influxdata/influxdb-client-go/internal/test"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"testing"
 )
 
 func TestAddDefaultTags(t *testing.T) {
@@ -50,3 +56,157 @@ func TestAddDefaultTags(t *testing.T) {
 
 	assert.Len(t, p.TagList(), 2)
 }
+
+func TestDefaultRetryDelay(t *testing.T) {
+	log.Log.SetDebugLevel(4)
+	hs := test.NewTestService(t, "http://localhost:9999")
+	opts := write.DefaultOptions()
+	ctx := context.Background()
+	srv := NewService("my-org", "my-bucket", hs, opts)
+	hs.SetReplyError(&ihttp.Error{
+		StatusCode: 502,
+	})
+	b1 := NewBatch("", opts.RetryInterval())
+	err := srv.HandleWrite(ctx, b1)
+	assert.NotNil(t, err)
+	assert.Equal(t, uint(5000), b1.retryDelay)
+	assert.Equal(t, 1, srv.retryQueue.list.Len())
+
+	time.Sleep(time.Millisecond * time.Duration(b1.retryDelay))
+	b2 := NewBatch("", opts.RetryInterval())
+	err = srv.HandleWrite(ctx, b2)
+	assert.NotNil(t, err)
+	assert.Equal(t, uint(25000), b1.retryDelay)
+	assert.Equal(t, 2, srv.retryQueue.list.Len())
+
+	time.Sleep(time.Millisecond * time.Duration(b1.retryDelay))
+	b3 := NewBatch("", opts.RetryInterval())
+	err = srv.HandleWrite(ctx, b3)
+	assert.NotNil(t, err)
+	assert.Equal(t, uint(125000), b1.retryDelay)
+	assert.Equal(t, 3, srv.retryQueue.list.Len())
+}
+
+func TestCustomRetryDelayWithFLush(t *testing.T) {
+	log.Log.SetDebugLevel(4)
+	hs := test.NewTestService(t, "http://localhost:9999")
+	opts := write.DefaultOptions().SetRetryInterval(1)
+	ctx := context.Background()
+	srv := NewService("my-org", "my-bucket", hs, opts)
+	hs.SetReplyError(&ihttp.Error{
+		StatusCode: 429,
+	})
+	b1 := NewBatch("1\n", opts.RetryInterval())
+	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.Sleep(time.Millisecond * time.Duration(b1.retryDelay))
+	b2 := NewBatch("2\n", opts.RetryInterval())
+	err = srv.HandleWrite(ctx, b2)
+	assert.NotNil(t, err)
+	assert.Equal(t, uint(5), b1.retryDelay)
+	assert.Equal(t, 2, srv.retryQueue.list.Len())
+
+	time.Sleep(time.Millisecond * time.Duration(b1.retryDelay))
+	b3 := NewBatch("3\n", opts.RetryInterval())
+	err = srv.HandleWrite(ctx, b3)
+	assert.NotNil(t, err)
+	assert.Equal(t, uint(25), b1.retryDelay)
+	assert.Equal(t, 3, srv.retryQueue.list.Len())
+
+	// let write pass and it will clear queue
+	time.Sleep(time.Millisecond * time.Duration(b1.retryDelay))
+	hs.SetReplyError(nil)
+	err = srv.HandleWrite(ctx, NewBatch("4\n", opts.RetryInterval()))
+	assert.Nil(t, err)
+	assert.Equal(t, 0, srv.retryQueue.list.Len())
+	require.Len(t, hs.Lines(), 4)
+	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])
+}
+
+func TestBufferOverwrite(t *testing.T) {
+	log.Log.SetDebugLevel(4)
+	hs := test.NewTestService(t, "http://localhost:9999")
+	//
+	opts := write.DefaultOptions().SetRetryInterval(1).SetRetryBufferLimit(15000)
+	ctx := context.Background()
+	srv := NewService("my-org", "my-bucket", hs, opts)
+	hs.SetReplyError(&ihttp.Error{
+		StatusCode: 429,
+	})
+	b1 := NewBatch("1\n", opts.RetryInterval())
+	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.Sleep(time.Millisecond * time.Duration(b1.retryDelay))
+	b2 := NewBatch("2\n", opts.RetryInterval())
+	err = srv.HandleWrite(ctx, b2)
+	assert.NotNil(t, err)
+	assert.Equal(t, uint(5), b1.retryDelay)
+	assert.Equal(t, 2, srv.retryQueue.list.Len())
+
+	time.Sleep(time.Millisecond * time.Duration(b1.retryDelay))
+	b3 := NewBatch("3\n", opts.RetryInterval())
+	err = srv.HandleWrite(ctx, b3)
+	assert.NotNil(t, err)
+	assert.Equal(t, uint(25), b1.retryDelay)
+	assert.Equal(t, 3, srv.retryQueue.list.Len())
+
+	// now it should drop b1
+	time.Sleep(time.Millisecond * time.Duration(b1.retryDelay))
+	b4 := NewBatch("4\n", opts.RetryInterval())
+	err = srv.HandleWrite(ctx, b4)
+	assert.NotNil(t, err)
+	assert.Equal(t, uint(1), b2.retryDelay)
+	assert.Equal(t, 3, srv.retryQueue.list.Len())
+
+	// let write pass and it will clear queue
+	time.Sleep(time.Millisecond * time.Duration(b1.retryDelay))
+	hs.SetReplyError(nil)
+	err = srv.HandleWrite(ctx, NewBatch("5\n", opts.RetryInterval()))
+	assert.Nil(t, err)
+	assert.Equal(t, 0, srv.retryQueue.list.Len())
+	require.Len(t, hs.Lines(), 4)
+	assert.Equal(t, "2", hs.Lines()[0])
+	assert.Equal(t, "3", hs.Lines()[1])
+	assert.Equal(t, "4", hs.Lines()[2])
+	assert.Equal(t, "5", hs.Lines()[3])
+}
+
+func TestMaxRetryInterval(t *testing.T) {
+	log.Log.SetDebugLevel(4)
+	hs := test.NewTestService(t, "http://localhost:9999")
+	//
+	opts := write.DefaultOptions().SetRetryInterval(1).SetMaxRetryInterval(10)
+	ctx := context.Background()
+	srv := NewService("my-org", "my-bucket", hs, opts)
+	hs.SetReplyError(&ihttp.Error{
+		StatusCode: 503,
+	})
+	b1 := NewBatch("1\n", opts.RetryInterval())
+	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.Sleep(time.Millisecond * time.Duration(b1.retryDelay))
+	b2 := NewBatch("2\n", opts.RetryInterval())
+	err = srv.HandleWrite(ctx, b2)
+	assert.NotNil(t, err)
+	assert.Equal(t, uint(5), b1.retryDelay)
+	assert.Equal(t, 2, srv.retryQueue.list.Len())
+
+	time.Sleep(time.Millisecond * time.Duration(b1.retryDelay))
+	b3 := NewBatch("3\n", opts.RetryInterval())
+	err = srv.HandleWrite(ctx, b3)
+	assert.NotNil(t, err)
+	assert.Equal(t, uint(10), b1.retryDelay)
+	assert.Equal(t, 3, srv.retryQueue.list.Len())
+}
diff --git a/options_test.go b/options_test.go
index 89d4abb..5be591c 100644
--- a/options_test.go
+++ b/options_test.go
@@ -20,9 +20,10 @@ func TestDefaultOptions(t *testing.T) {
 	assert.Equal(t, false, opts.UseGZip())
 	assert.Equal(t, uint(1000), opts.FlushInterval())
 	assert.Equal(t, time.Nanosecond, opts.Precision())
-	assert.Equal(t, uint(10000), opts.RetryBufferLimit())
-	assert.Equal(t, uint(1000), opts.RetryInterval())
+	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())
@@ -38,7 +39,8 @@ func TestSettingsOptions(t *testing.T) {
 		SetFlushInterval(5000).
 		SetPrecision(time.Millisecond).
 		SetRetryBufferLimit(5).
-		SetRetryInterval(5000).
+		SetRetryInterval(1000).
+		SetMaxRetryInterval(10000).
 		SetMaxRetries(7).
 		SetTLSConfig(tlsConfig).
 		SetHTTPRequestTimeout(50).
@@ -49,7 +51,8 @@ func TestSettingsOptions(t *testing.T) {
 	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(5000), opts.RetryInterval())
+	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())