OnError callback in esutil.BulkIndexerConfig is invoked duplicately

[iamorchid]

In bulk_indexer.go, I see logic below:

func (w *worker) run() {
	go func() {     
                                ...  
                                if err := w.flush(ctx); err != nil {
					w.mu.Unlock()
					if w.bi.config.OnError != nil {
						w.bi.config.OnError(ctx, err)
					}
					continue
				}
                                 ...
        }()
}
func (w *worker) flush(ctx context.Context) error {
...
	if res.IsError() {
		atomic.AddUint64(&w.bi.stats.numFailed, uint64(len(w.items)))
		// TODO(karmi): Wrap error (include response struct)
		if w.bi.config.OnError != nil {
			w.bi.config.OnError(ctx, fmt.Errorf("flush: %s", err))
		}
		return fmt.Errorf("flush: %s", res.String())
	}
...
}

When flush fails, bi.config.OnError is called twice. One happens in flush function, and the other in the run function.

[ljluestc]

package main

import (
	"context"
	"fmt"
	"sync"
	"sync/atomic"
)

// bulkIndexerConfig holds configuration for the bulk indexer.
type bulkIndexerConfig struct {
	OnError func(context.Context, error) // Callback for error handling
}

// bulkIndexerStats tracks indexing statistics.
type bulkIndexerStats struct {
	numFailed uint64 // Number of failed items
}

// bulkIndexer manages bulk indexing operations.
type bulkIndexer struct {
	config bulkIndexerConfig
	stats  bulkIndexerStats
}

// worker handles a subset of bulk indexing tasks.
type worker struct {
	bi    *bulkIndexer
	items []interface{} // Items to be indexed
	mu    sync.Mutex   // Protects worker state
}

// run starts the worker's main loop, processing items and flushing them.
func (w *worker) run(ctx context.Context) {
	go func() {
		for {
			select {
			case <-ctx.Done():
				w.mu.Lock()
				if len(w.items) > 0 {
					_ = w.flush(ctx) // Errors handled in flush
				}
				w.mu.Unlock()
				return
			default:
				w.mu.Lock()
				if len(w.items) == 0 {
					w.mu.Unlock()
					continue
				}
				if err := w.flush(ctx); err != nil {
					// Errors (including OnError) handled in flush; just continue
					w.mu.Unlock()
					continue
				}
				w.items = w.items[:0] // Clear items after successful flush
				w.mu.Unlock()
			}
		}
	}()
}

// flush sends accumulated items to the indexing service.
func (w *worker) flush(ctx context.Context) error {
	if len(w.items) == 0 {
		return nil
	}

	// Simulate indexing request (e.g., to Elasticsearch).
	res, err := simulateIndexRequest(w.items)
	if err != nil {
		atomic.AddUint64(&w.bi.stats.numFailed, uint64(len(w.items)))
		wrappedErr := fmt.Errorf("flush: %v", err)
		if w.bi.config.OnError != nil {
			w.bi.config.OnError(ctx, wrappedErr)
		}
		return wrappedErr
	}

	if res.IsError() {
		atomic.AddUint64(&w.bi.stats.numFailed, uint64(len(w.items)))
		err := fmt.Errorf("flush: %s", res.String())
		if w.bi.config.OnError != nil {
			w.bi.config.OnError(ctx, err)
		}
		return err
	}

	// Success: update stats or other tasks
	return nil
}

// response simulates an indexing service response.
type response struct {
	status string
}

func (r *response) IsError() bool {
	return r.status != "OK"
}

func (r *response) String() string {
	return fmt.Sprintf("status=%s", r.status)
}

// simulateIndexRequest mocks an indexing request.
func simulateIndexRequest(items []interface{}) (*response, error) {
	// Simulate a failed response for demonstration
	return &response{status: "ERROR"}, nil
}

// Example usage
func main() {
	config := bulkIndexerConfig{
		OnError: func(ctx context.Context, err error) {
			fmt.Printf("Error: %v\n", err)
		},
	}
	bi := &bulkIndexer{
		config: config,
	}
	w := &worker{
		bi:    bi,
		items: []interface{}{"doc1", "doc2"},
	}

	ctx := context.Background()
	w.run(ctx)
}