Notify items if an error occurs in bulk indexer

esutil/bulk_indexer.go

@@ -573,6 +573,7 @@ func (w *worker) flush(ctx context.Context) error {
 		if w.bi.config.OnError != nil {
 			w.bi.config.OnError(ctx, fmt.Errorf("flush: %s", err))
 		}
+		w.notifyItemsOnError(ctx, err)
 		return fmt.Errorf("flush: %s", err)
 	}
 	if res.Body != nil {
@@ -584,6 +585,7 @@ func (w *worker) flush(ctx context.Context) error {
 		if w.bi.config.OnError != nil {
 			w.bi.config.OnError(ctx, fmt.Errorf("flush: %s", res.String()))
 		}
+		w.notifyItemsOnError(ctx, err)
 		return fmt.Errorf("flush: %s", res.String())
 	}
 
@@ -592,6 +594,7 @@ func (w *worker) flush(ctx context.Context) error {
 		if w.bi.config.OnError != nil {
 			w.bi.config.OnError(ctx, fmt.Errorf("flush: %s", err))
 		}
+		w.notifyItemsOnError(ctx, err)
 		return fmt.Errorf("flush: error parsing response body: %s", err)
 	}
 
@@ -638,6 +641,14 @@ func (w *worker) flush(ctx context.Context) error {
 	return err
 }
 
+func (w *worker) notifyItemsOnError(ctx context.Context, err error) {
+	for _, item := range w.items {
+		if item.OnFailure != nil {
+			item.OnFailure(ctx, item, BulkIndexerResponseItem{}, fmt.Errorf("flush: %w", err))
+		}
+	}
+}
+
 type defaultJSONDecoder struct{}
 
 func (d defaultJSONDecoder) UnmarshalFromReader(r io.Reader, blk *BulkIndexerResponse) error {