Repository: influxdata/influxdb-client-go (go)

Issue:
Does not retry on connection reset
__Steps to reproduce:__
List the minimal actions needed to reproduce the behavior.

1. Don't start the influx server, so the connection for writing data is refused
2. Configure the Client to use retries
3. Try to write data
4. Turn on the influx server before the retry period is exceeded
5. See whether the data that should be written somewhen during retries is written

__Expected behavior:__
The data is written to the server, if the server is available within the retry period

__Actual behavior:__
The client stops retrying after the first connection reset

__Specifications:__
 - Client Version: v2.2.0
 - InfluxDB Version: 1.8.3
 - Platform: linux - arm


Pull request description:
fix: Retry also on network errors
Closes #226 

## Proposed Changes

Retry write also on network errors

## 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 a518436..8e8e254 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,8 @@
+## 2.2.2 [in progress]
+### Bug fixes
+1. [#229](https://github.com/influxdata/influxdb-client-go/pull/229) Connection errors are also subject for retrying.
+
+
 ## 2.2.1 [2020-12-24]
 ### Bug fixes
 1. [#220](https://github.com/influxdata/influxdb-client-go/pull/220) Fixed runtime error occurring when calling v2 API on v1 server.
diff --git a/api/http/service.go b/api/http/service.go
index 8801d1c..db22de5 100644
--- a/api/http/service.go
+++ b/api/http/service.go
@@ -111,7 +111,7 @@ func (s *service) DoHTTPRequest(req *http.Request, requestCallback RequestCallba
 	}
 
 	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
-		return s.handleHTTPError(resp)
+		return s.parseHTTPError(resp)
 	}
 	if responseCallback != nil {
 		err := responseCallback(resp)
@@ -134,7 +134,7 @@ func (s *service) DoHTTPRequestWithResponse(req *http.Request, requestCallback R
 	return s.client.Do(req)
 }
 
-func (s *service) handleHTTPError(r *http.Response) *Error {
+func (s *service) parseHTTPError(r *http.Response) *Error {
 	// successful status code range
 	if r.StatusCode >= 200 && r.StatusCode < 300 {
 		return nil
diff --git a/api/write.go b/api/write.go
index dfc2dc8..505b623 100644
--- a/api/write.go
+++ b/api/write.go
@@ -95,7 +95,7 @@ func (w *WriteAPIImpl) waitForFlushing() {
 			break
 		}
 		log.Info("Waiting buffer is flushed")
-		time.Sleep(time.Millisecond)
+		<-time.After(time.Millisecond)
 	}
 	for {
 		w.writeInfoCh <- writeBuffInfoReq{}
@@ -104,7 +104,7 @@ func (w *WriteAPIImpl) waitForFlushing() {
 			break
 		}
 		log.Info("Waiting buffer is flushed")
-		time.Sleep(time.Millisecond)
+		<-time.After(time.Millisecond)
 	}
 }
 
diff --git a/internal/write/service.go b/internal/write/service.go
index 9b6410c..75793b8 100644
--- a/internal/write/service.go
+++ b/internal/write/service.go
@@ -99,7 +99,7 @@ func (w *Service) HandleWrite(ctx context.Context, batch *Batch) error {
 		if batchToWrite != nil {
 			perror := w.WriteBatch(ctx, batchToWrite)
 			if perror != nil {
-				if perror.StatusCode >= http.StatusTooManyRequests {
+				if perror.StatusCode == 0 || perror.StatusCode >= http.StatusTooManyRequests {
 					log.Errorf("Write error: %s\nBatch kept for retrying\n", perror.Error())
 					if perror.RetryAfter > 0 {
 						batchToWrite.retryDelay = perror.RetryAfter * 1000


Test patch:
diff --git a/api/query_test.go b/api/query_test.go
index f1f8a44..4ed20f0 100644
--- a/api/query_test.go
+++ b/api/query_test.go
@@ -483,7 +483,7 @@ func TestQueryRawResult(t *testing.T) {
 	csvTable := strings.Join(csvRows, "\r\n")
 	csvTable = fmt.Sprintf("%s\r\n", csvTable)
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		time.Sleep(100 * time.Millisecond)
+		<-time.After(100 * time.Millisecond)
 		if r.Method == http.MethodPost {
 			rbody, _ := ioutil.ReadAll(r.Body)
 			fmt.Printf("Req: %s\n", string(rbody))
@@ -893,7 +893,7 @@ func TestEmptyValue(t *testing.T) {
 
 func TestFluxError(t *testing.T) {
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		time.Sleep(100 * time.Millisecond)
+		<-time.After(100 * time.Millisecond)
 		if r.Method == http.MethodPost {
 			_, _ = ioutil.ReadAll(r.Body)
 			w.Header().Set("Content-Type", "application/json")
diff --git a/api/tasks_e2e_test.go b/api/tasks_e2e_test.go
index 3bf5b45..621afe9 100644
--- a/api/tasks_e2e_test.go
+++ b/api/tasks_e2e_test.go
@@ -390,7 +390,7 @@ func TestTasksAPI_Runs(t *testing.T) {
 	require.Nil(t, err)
 	require.NotNil(t, task)
 	//wait for task to run 10x
-	time.Sleep(time.Second * 10)
+	<-time.After(time.Second * 10)
 
 	logs, err := tasksAPI.FindLogs(ctx, task)
 	require.Nil(t, err)
@@ -431,7 +431,7 @@ func TestTasksAPI_Runs(t *testing.T) {
 	require.Nil(t, err)
 	require.NotNil(t, task)
 	//wait for task to start and be running
-	time.Sleep(1500 * time.Millisecond)
+	<-time.After(1500 * time.Millisecond)
 
 	// we should get a running run
 	runs, err = tasksAPI.FindRuns(ctx, task, nil)
diff --git a/api/writeAPIBlocking_test.go b/api/writeAPIBlocking_test.go
index 407903b..9884c39 100644
--- a/api/writeAPIBlocking_test.go
+++ b/api/writeAPIBlocking_test.go
@@ -63,7 +63,7 @@ func TestWriteContextCancel(t *testing.T) {
 	var wg sync.WaitGroup
 	wg.Add(1)
 	go func() {
-		time.Sleep(time.Second)
+		<-time.After(time.Second)
 		err = writeAPI.WriteRecord(ctx, lines...)
 		wg.Done()
 	}()
diff --git a/api/write_test.go b/api/write_test.go
index 12ee63c..19c6923 100644
--- a/api/write_test.go
+++ b/api/write_test.go
@@ -133,7 +133,7 @@ func TestFlushInterval(t *testing.T) {
 		writeAPI.WritePoint(p)
 	}
 	require.Len(t, service.Lines(), 0)
-	time.Sleep(time.Millisecond * 600)
+	<-time.After(time.Millisecond * 600)
 	require.Len(t, service.Lines(), 5)
 	writeAPI.Close()
 
@@ -143,7 +143,7 @@ func TestFlushInterval(t *testing.T) {
 		writeAPI.WritePoint(p)
 	}
 	require.Len(t, service.Lines(), 0)
-	time.Sleep(time.Millisecond * 2100)
+	<-time.After(time.Millisecond * 2100)
 	require.Len(t, service.Lines(), 5)
 
 	writeAPI.Close()
@@ -175,7 +175,7 @@ func TestRetry(t *testing.T) {
 	}
 	writeAPI.waitForFlushing()
 	require.Len(t, service.Lines(), 0)
-	time.Sleep(5*time.Second + 50*time.Millisecond)
+	<-time.After(5*time.Second + 50*time.Millisecond)
 	for i := 10; i < 15; i++ {
 		writeAPI.WritePoint(points[i])
 	}
diff --git a/client_test.go b/client_test.go
index 3d0be17..dcc8a9a 100644
--- a/client_test.go
+++ b/client_test.go
@@ -79,7 +79,7 @@ func TestWriteAPIManagement(t *testing.T) {
 
 func TestUserAgent(t *testing.T) {
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		time.Sleep(100 * time.Millisecond)
+		<-time.After(100 * time.Millisecond)
 		if r.Header.Get("User-Agent") == http2.UserAgent {
 			w.WriteHeader(http.StatusOK)
 		} else {
@@ -99,7 +99,7 @@ func TestUserAgent(t *testing.T) {
 
 func TestServerError429(t *testing.T) {
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		time.Sleep(100 * time.Millisecond)
+		<-time.After(100 * time.Millisecond)
 		w.Header().Set("Retry-After", "1")
 		w.Header().Set("Content-Type", "application/json")
 		w.WriteHeader(http.StatusTooManyRequests)
@@ -135,7 +135,7 @@ func TestServerOnPath(t *testing.T) {
 
 func TestServerErrorNonJSON(t *testing.T) {
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		time.Sleep(100 * time.Millisecond)
+		<-time.After(100 * time.Millisecond)
 		w.WriteHeader(http.StatusInternalServerError)
 		_, _ = w.Write([]byte(`internal server error`))
 	}))
diff --git a/internal/write/service_test.go b/internal/write/service_test.go
index 75748a5..6f1e9b4 100644
--- a/internal/write/service_test.go
+++ b/internal/write/service_test.go
@@ -6,6 +6,7 @@ package write
 
 import (
 	"context"
+	"errors"
 	"testing"
 	"time"
 
@@ -72,14 +73,14 @@ func TestDefaultRetryDelay(t *testing.T) {
 	assert.Equal(t, uint(5000), b1.retryDelay)
 	assert.Equal(t, 1, srv.retryQueue.list.Len())
 	//wait retry delay + little more
-	time.Sleep(time.Millisecond*time.Duration(b1.retryDelay) + time.Microsecond*5)
+	<-time.After(time.Millisecond*time.Duration(b1.retryDelay) + time.Microsecond*5)
 	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) + time.Microsecond*5)
+	<-time.After(time.Millisecond*time.Duration(b1.retryDelay) + time.Microsecond*5)
 	b3 := NewBatch("", opts.RetryInterval())
 	err = srv.HandleWrite(ctx, b3)
 	assert.NotNil(t, err)
@@ -103,7 +104,7 @@ func TestCustomRetryDelayWithFLush(t *testing.T) {
 	assert.Equal(t, 1, srv.retryQueue.list.Len())
 
 	//wait retry delay + little more
-	time.Sleep(time.Millisecond*time.Duration(b1.retryDelay) + time.Microsecond*5)
+	<-time.After(time.Millisecond*time.Duration(b1.retryDelay) + time.Microsecond*5)
 	b2 := NewBatch("2\n", opts.RetryInterval())
 	err = srv.HandleWrite(ctx, b2)
 	assert.NotNil(t, err)
@@ -111,7 +112,7 @@ func TestCustomRetryDelayWithFLush(t *testing.T) {
 	assert.Equal(t, 2, srv.retryQueue.list.Len())
 
 	//wait retry delay + little more
-	time.Sleep(time.Millisecond*time.Duration(b1.retryDelay) + time.Microsecond*5)
+	<-time.After(time.Millisecond*time.Duration(b1.retryDelay) + time.Microsecond*5)
 	b3 := NewBatch("3\n", opts.RetryInterval())
 	err = srv.HandleWrite(ctx, b3)
 	assert.NotNil(t, err)
@@ -119,7 +120,7 @@ func TestCustomRetryDelayWithFLush(t *testing.T) {
 	assert.Equal(t, 3, srv.retryQueue.list.Len())
 
 	// let write pass and it will clear queue
-	time.Sleep(time.Millisecond*time.Duration(b1.retryDelay) + time.Microsecond*5)
+	<-time.After(time.Millisecond*time.Duration(b1.retryDelay) + time.Microsecond*5)
 	hs.SetReplyError(nil)
 	err = srv.HandleWrite(ctx, NewBatch("4\n", opts.RetryInterval()))
 	assert.Nil(t, err)
@@ -147,14 +148,14 @@ func TestBufferOverwrite(t *testing.T) {
 	assert.Equal(t, uint(1), b1.retryDelay)
 	assert.Equal(t, 1, srv.retryQueue.list.Len())
 
-	time.Sleep(time.Millisecond * time.Duration(b1.retryDelay))
+	<-time.After(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))
+	<-time.After(time.Millisecond * time.Duration(b1.retryDelay))
 	b3 := NewBatch("3\n", opts.RetryInterval())
 	err = srv.HandleWrite(ctx, b3)
 	assert.NotNil(t, err)
@@ -162,7 +163,7 @@ func TestBufferOverwrite(t *testing.T) {
 	assert.Equal(t, 3, srv.retryQueue.list.Len())
 
 	// now it should drop b1
-	time.Sleep(time.Millisecond * time.Duration(b1.retryDelay))
+	<-time.After(time.Millisecond * time.Duration(b1.retryDelay))
 	b4 := NewBatch("4\n", opts.RetryInterval())
 	err = srv.HandleWrite(ctx, b4)
 	assert.NotNil(t, err)
@@ -170,7 +171,7 @@ func TestBufferOverwrite(t *testing.T) {
 	assert.Equal(t, 3, srv.retryQueue.list.Len())
 
 	// let write pass and it will clear queue
-	time.Sleep(time.Millisecond * time.Duration(b1.retryDelay))
+	<-time.After(time.Millisecond * time.Duration(b1.retryDelay))
 	hs.SetReplyError(nil)
 	err = srv.HandleWrite(ctx, NewBatch("5\n", opts.RetryInterval()))
 	assert.Nil(t, err)
@@ -198,17 +199,63 @@ func TestMaxRetryInterval(t *testing.T) {
 	assert.Equal(t, uint(1), b1.retryDelay)
 	assert.Equal(t, 1, srv.retryQueue.list.Len())
 
-	time.Sleep(time.Millisecond * time.Duration(b1.retryDelay))
+	<-time.After(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))
+	<-time.After(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())
 }
+
+func TestRetryOnConnectionError(t *testing.T) {
+	log.Log.SetLogLevel(log.DebugLevel)
+	hs := test.NewTestService(t, "http://localhost:8086")
+	//
+	opts := write.DefaultOptions().SetRetryInterval(1).SetRetryBufferLimit(15000)
+	ctx := context.Background()
+	srv := NewService("my-org", "my-bucket", hs, opts)
+
+	hs.SetReplyError(&http.Error{
+		Err: errors.New("connection refused"),
+	})
+
+	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.After(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.After(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.After(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])
+
+}
diff --git a/options_test.go b/options_test.go
index 0c62341..9c61ac3 100644
--- a/options_test.go
+++ b/options_test.go
@@ -76,11 +76,11 @@ func TestTimeout(t *testing.T) {
 		,,0,2020-02-17T22:19:49.747562847Z,2020-02-18T22:19:49.747562847Z,2020-02-18T22:08:44.850214724Z,6.6,f,test,1,adsfasdf
 		`
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		time.Sleep(100 * time.Millisecond)
+		<-time.After(100 * time.Millisecond)
 		if r.Method == http.MethodPost {
 			w.Header().Set("Content-Type", "text/csv")
 			w.WriteHeader(http.StatusOK)
-			time.Sleep(2 * time.Second)
+			<-time.After(2 * time.Second)
 			_, err := w.Write([]byte(response))
 			if err != nil {
 				w.WriteHeader(http.StatusInternalServerError)