Handle error of ES bulk request

elasticsearch/bulk/bulk.go

@@ -2,9 +2,14 @@ package bulk
 
 import (
 	"bytes"
+	"encoding/json"
+	"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"
@@ -180,11 +185,47 @@ func (b *Bulk) flushMessages() error {
 func (b *Bulk) bulkRequest() error {
 	startedTime := time.Now()
 	reader := bytes.NewReader(b.batch)
-	_, err := b.esClient.Bulk(reader)
+	resp, err := b.esClient.Bulk(reader)
 	b.metric.BulkRequestProcessLatencyMs = time.Since(startedTime).Milliseconds()
 	if err != nil {
 		return err
 	}
+	err = b.handleResp(resp)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (b *Bulk) handleResp(resp *esapi.Response) error {
+	if resp == nil {
+		return fmt.Errorf("ERROR occurred when bulk request. Error: Response is nil")
+	}
+	if resp.IsError() {
+		return fmt.Errorf("ERROR occurred when bulk request. Error:%s", resp.String())
+	}
+	bodyBuffer := new(bytes.Buffer)
+	_, err := bodyBuffer.ReadFrom(resp.Body)
+	if err != nil {
+		return err
+	}
+	bodyInterface := make(map[string]interface{})
+	err = json.Unmarshal(bodyBuffer.Bytes(), &bodyInterface)
+	if err != nil {
+		return err
+	} else if bodyInterface["errors"] != nil {
+		if hasError, ok := bodyInterface["errors"].(bool); ok && hasError {
+			var sb strings.Builder
+			sb.WriteString("ERROR occurred when bulk request. Items will be listed below:\n")
+			items := bodyInterface["items"]
+			if items != nil {
+				for _, v := range items.([]interface{}) {
+					sb.WriteString(fmt.Sprintf("Item: %v\n", v))
+				}
+			}
+			return fmt.Errorf(sb.String())
+		}
+	}
 	return nil
 }