title: Go elasticsearch insert bulk data from incoming pulsar
body: <p>I have to use goelastic library inserting the datas bulkly from coming pulsar. But i have a problem.</p>
  <p>Firstly, pulsar send 1000 datas per partial bulkly. Then when i insert the elastic, there are a problem sometimes. This problem is attached. This problem cause data loss. Thanks for answer...
  <code>ERROR: circuit_breaking_exception: [parent] Data too large, data for [indices:data/write/bulk[s]] would be [524374312/500mb], which is larger than the limit of [510027366/486.3mb], real usage: [524323448/500mb], new bytes reserved: [50864/49.6kb], usages [request=0/0b, fielddata=160771183/153.3mb, in_flight_requests=50864/49.6kb, model_inference=0/0b, eql_sequence=0/0b, accounting=6898128/6.5mb]</code></p>
  <p>This section is bulk code.</p>
  <pre><code>
  func InsertElastic(y []models.CP, ElasticStruct *config.ElasticStruct) {
      fmt.Println(&quot;------------------&quot;)
      bi, err := esutil.NewBulkIndexer(esutil.BulkIndexerConfig{
          Index:      enum.IndexName,
          Client:     ElasticStruct.Client,
          FlushBytes: 10e+6,
      })
      if err != nil {
          panic(err)
      }
      start := time.Now().UTC()
  
      for _, x := range y {
          data, err := json.Marshal(x)
          if err != nil {
              panic(err)
          }
          err = bi.Add(
              context.Background(),
              esutil.BulkIndexerItem{
                  Action: &quot;index&quot;,
  
                  Body: bytes.NewReader(data),
  
                  OnSuccess: func(ctx context.Context, item esutil.BulkIndexerItem, res esutil.BulkIndexerResponseItem) {
                      i++
                  },
  
                  OnFailure: func(ctx context.Context, item esutil.BulkIndexerItem, res esutil.BulkIndexerResponseItem, err error) {
                      if err != nil {
                          log.Printf(&quot;ERROR: %s&quot;, err)
                      } else {
                          log.Printf(&quot;ERROR: %s: %s&quot;, res.Error.Type, res.Error.Reason)
                      }
                  },
              },
          )
          if err != nil {
              log.Fatalf(&quot;Unexpected error: %s&quot;, err)
          }
          x++
  
      }
      if err := bi.Close(context.Background()); err != nil {
          log.Fatalf(&quot;Unexpected error: %s&quot;, err)
      }
      dur := time.Since(start)
      fmt.Println(dur)
      fmt.Println(&quot;Success writing data to elastic : &quot;, i)
      fmt.Println(&quot;Success incoming data from pulsar : &quot;, x)
      fmt.Println(&quot;Difference : &quot;, x-i)
      fmt.Println(&quot;Now : &quot;, time.Now().UTC().String())
      if i &lt; x {
          fmt.Println(&quot;FATAL&quot;)
      }
      fmt.Println(&quot;------------------&quot;)
  
  }
  </code></pre>