I am trying to insert bulk data. I am not getting any error except for in response property error: true. Here is my code snippet.

func BulkInsert(ctx context.Context, products []model.Pt) error {
    Counter  
    // create new bulk processor from client
    bulkProcessor, err := ElasticVar.BulkProcessor().
        Name("Bulk Worker 1").
        Workers(5).
        BulkActions(1000).
        BulkSize(2 << 20).
        FlushInterval(1 * time.Second).
        Stats(true).
        After(after).
        Do(context.Background())
    if err != nil {
        log.Println("NewBulkProcessorService err", err)
        return err
    }
    defer bulkProcessor.Close()
    storeIds := map[string]string{}
    // Enqueue the document
    for _, product := range products {
        if _, ok := storeIds[product.Id]; ok {
            continue
        } else {
            storeIds[product.StoreId] = product.StoreId
        }

        dataJSON, err := json.Marshal(product)
        if err != nil {
            // Look up the failed documents with res.Failed(), and e.g. recommit
            log.Println("elastic marshal failed", err)
            return errors.New("elastic marshal failed")
        }
        bulkProcessor.Add(elastic.NewBulkIndexRequest().
            OpType("index").
            Index(Index).
            Type(Type).
            Id(product.Id).
            Doc(string(dataJSON)))
    }
    stats := bulkProcessor.Stats()
    fmt.Printf("Number of times flush has been invoked: %d\n", stats.Flushed)
    fmt.Printf("Number of times workers committed reqs: %d\n", stats.Committed)
    fmt.Printf("Number of requests indexed            : %d\n", stats.Indexed)
    fmt.Printf("Number of requests reported as created: %d\n", stats.Created)
    fmt.Printf("Number of requests reported as updated: %d\n", stats.Updated)
    fmt.Printf("Number of requests reported as success: %d\n", stats.Succeeded)
    fmt.Printf("Number of requests reported as failed : %d\n", stats.Failed)
    fmt.Printf("\n\n")
    for i, w := range stats.Workers {
        fmt.Printf("Worker %d: Number of requests queued: %d\n", i, w.Queued)
        fmt.Printf("           Last response time       : %v\n", w.LastDuration)
    }
    fmt.Printf(`Inserted Count: %d`, Counter)
    return nil
}

func after(executionID int64, requests []elastic.BulkableRequest, response *elastic.BulkResponse, err error) {
    if err != nil {
        log.Printf("bulk commit failed, err: %v\n", err)
    }
    // do what ever you want in case bulk commit success
    log.Printf("commit response=%v\n", response.Errors)
    log.Printf("commit successfully, len(requests)=%d\n", len(requests))
}

I am adding all the data in bulk.

In the callback after the response. error is true. When I query the data. None of the records is inserted. Please help.

CodePudding user response：

I was able to debug it by access the error string through created mether

for _, info := range response.Created() {
    fmt.Println("nBulk response created error:", info.Error)
    fmt.Println("nBulk response created: error reason4", info.Error.Reason)
    fmt.Println("nBulk response IndexProduct:", info.IndexProduct)
}

The problem was the I was passing Type = "product" instead of Type = "doc"