Handle error of ES bulk request

elasticsearch/bulk/bulk.go

@@ -2,9 +2,13 @@ package bulk
 
 import (
 	"bytes"
+	"fmt"
+	"strings"
 	"sync"
 	"time"
 
+	"github.com/elastic/go-elasticsearch/v7/esapi"
+
 	"github.com/Trendyol/go-dcp-client/logger"
 
 	"github.com/Trendyol/go-dcp-client/models"
@@ -13,6 +17,7 @@ import (
 	"github.com/Trendyol/go-elasticsearch-connect-couchbase/elasticsearch/document"
 	"github.com/Trendyol/go-elasticsearch-connect-couchbase/helper"
 	"github.com/elastic/go-elasticsearch/v7"
+	"github.com/json-iterator/go"
 )
 
 type Bulk struct {
@@ -180,14 +185,61 @@ func (b *Bulk) flushMessages() error {
 func (b *Bulk) bulkRequest() error {
 	startedTime := time.Now()
 	reader := bytes.NewReader(b.batch)
-	_, err := b.esClient.Bulk(reader)
+	r, err := b.esClient.Bulk(reader)
 	b.metric.BulkRequestProcessLatencyMs = time.Since(startedTime).Milliseconds()
 	if err != nil {
 		return err
 	}
+	err = hasResponseError(r)
+	if err != nil {
+		return err
+	}
 	return nil
 }
 
 func (b *Bulk) GetMetric() *Metric {
 	return b.metric
 }
+
+func hasResponseError(r *esapi.Response) error {
+	if r == nil {
+		return fmt.Errorf("esapi response is nil")
+	}
+	if r.IsError() {
+		return fmt.Errorf("bulk request has error %v", r.String())
+	}
+	rb := new(bytes.Buffer)
+	_, err := rb.ReadFrom(r.Body)
+	if err != nil {
+		return err
+	}
+	b := make(map[string]interface{})
+	err = jsoniter.Unmarshal(rb.Bytes(), &b)
+	if err != nil {
+		return err
+	}
+	return checkErrorsIsTrue(b)
+}
+
+func checkErrorsIsTrue(body map[string]interface{}) error {
+	if hasError, ok := body["errors"].(bool); ok && hasError {
+		var sb strings.Builder
+		sb.WriteString("bulk request has error. Errors will be listed below:\n")
+		if items, ok := body["items"].([]interface{}); ok {
+			for _, i := range items {
+				if item, ok := i.(map[string]interface{}); ok {
+					for _, v := range item {
+						if iv, ok := v.(map[string]interface{}); ok {
+							if iv["error"] != nil {
+								sb.WriteString(fmt.Sprintf("%v\n", i))
+							}
+						}
+						break
+					}
+				}
+			}
+		}
+		return fmt.Errorf(sb.String())
+	}
+	return nil
+}

go.mod

@@ -5,6 +5,7 @@ go 1.20
 require (
 	github.com/Trendyol/go-dcp-client v0.0.59
 	github.com/elastic/go-elasticsearch/v7 v7.17.7
+	github.com/json-iterator/go v1.1.12
 	github.com/prometheus/client_golang v1.15.1
 	github.com/valyala/fasthttp v1.47.0
 	gopkg.in/yaml.v3 v3.0.1
@@ -33,7 +34,6 @@ require (
 	github.com/google/gofuzz v1.2.0 // indirect
 	github.com/google/uuid v1.3.0 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
-	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/klauspost/compress v1.16.3 // indirect
 	github.com/mailru/easyjson v0.7.7 // indirect
 	github.com/mattn/go-colorable v0.1.13 // indirect