fatal error: concurrent map read and map write

[ppg-coder]

I use github.com/olivere/elastic/v7 v7.0.22 and go 1.15

I use bulk processor to batch index and update doc, my setup code like this

func (i *SpuEs) BulkProcessor(ctx context.Context) (*elastic.BulkProcessor, error) {
	return i.EsClient.BulkProcessor().
		Name(i.EsIndexName).
		Workers(5).
		BulkActions(4096).
		BulkSize(2 << 20).
		FlushInterval(2 * time.Second).
		After(func(executionId int64, requests []elastic.BulkableRequest, response *elastic.BulkResponse, err error) {
			if err != nil {
				failedNum := 0
				if response != nil {
					failedNum = len(response.Failed())
				}
				xlog.WithEvent(i.EsIndexName).Errorf("BulkProcessor failed, id:%v, req_num:%v, failed_num:%v, err:%v", executionId, len(requests), failedNum, err)
				return
			}
			if response == nil || !response.Errors {
				return
			}
			for _, item := range response.Items {
				for key, val := range item {
					if val.Error == nil {
						continue
					}
					xlog.WithEvent(i.EsIndexName).Errorf("BulkProcessor failed, key:%v, res:%v, err:%v", key, val.Result, val.Error.Reason)
				}
			}
		}).
		Do(ctx)
}

atfer my service start, the fatal error occurred with probability

fatal error: concurrent map read and map write

goroutine 455 [running]:
runtime.throw(0x1d408db, 0x21)
	/usr/local/go/src/runtime/panic.go:1116 +0x72 fp=0xc0034599c8 sp=0xc003459998 pc=0x437cb2
runtime.mapaccess2(0x1a1ac80, 0xc005cb8ff0, 0xc00aea08c0, 0xc00aea08c0, 0xc0055fe160)
	/usr/local/go/src/runtime/map.go:469 +0x25b fp=0xc003459a08 sp=0xc0034599c8 pc=0x40f91b
reflect.mapaccess(0x1a1ac80, 0xc005cb8ff0, 0xc00aea08c0, 0x1d2b0ae)
	/usr/local/go/src/runtime/map.go:1309 +0x3f fp=0xc003459a40 sp=0xc003459a08 pc=0x469d5f
reflect.Value.MapIndex(0x1a1ac80, 0xc005cb8ff0, 0x15, 0x196f860, 0xc00aea08c0, 0x98, 0x19e70e0, 0xc00aea09c0, 0x94)
	/usr/local/go/src/reflect/value.go:1189 +0x16e fp=0xc003459ab8 sp=0xc003459a40 pc=0x4a3d0e
encoding/json.mapEncoder.encode(0x1d88598, 0xc006186100, 0x1a1ac80, 0xc005cb8ff0, 0x15, 0x1a10100)
	/usr/local/go/src/encoding/json/encode.go:801 +0x30d fp=0xc003459c30 sp=0xc003459ab8 pc=0x75bdad
encoding/json.mapEncoder.encode-fm(0xc006186100, 0x1a1ac80, 0xc005cb8ff0, 0x15, 0x2bd0100)
	/usr/local/go/src/encoding/json/encode.go:777 +0x65 fp=0xc003459c70 sp=0xc003459c30 pc=0x768405
encoding/json.(*encodeState).reflectValue(0xc006186100, 0x1a1ac80, 0xc005cb8ff0, 0x15, 0xc003450100)
	/usr/local/go/src/encoding/json/encode.go:358 +0x82 fp=0xc003459ca8 sp=0xc003459c70 pc=0x758f42
encoding/json.(*encodeState).marshal(0xc006186100, 0x1a1ac80, 0xc005cb8ff0, 0x100, 0x0, 0x0)
	/usr/local/go/src/encoding/json/encode.go:330 +0xf4 fp=0xc003459d08 sp=0xc003459ca8 pc=0x758b34
encoding/json.Marshal(0x1a1ac80, 0xc005cb8ff0, 0x45, 0xc000f29950, 0x45, 0x0, 0x0)
	/usr/local/go/src/encoding/json/encode.go:161 +0x52 fp=0xc003459d80 sp=0xc003459d08 pc=0x757fb2
github.com/olivere/elastic/v7.(*BulkIndexRequest).Source(0xc006bbaee0, 0xc003459ee0, 0x2, 0x0, 0x1, 0x1)
	/go/pkg/mod/github.com/olivere/elastic/v7@v7.0.22/bulk_index_request.go:240 +0x39a fp=0xc003459e68 sp=0xc003459d80 pc=0x147efba
github.com/olivere/elastic/v7.(*bulkWorker).work(0xc001994b80, 0x1f72580, 0xc0003c1a80)
	/go/pkg/mod/github.com/olivere/elastic/v7@v7.0.22/bulk_processor.go:484 +0x46b fp=0xc003459fc8 sp=0xc003459e68 pc=0x148260b
runtime.goexit()
	/usr/local/go/src/runtime/asm_amd64.s:1374 +0x1 fp=0xc003459fd0 sp=0xc003459fc8 pc=0x4707e1
created by github.com/olivere/elastic/v7.(*BulkProcessor).Start
	/go/pkg/mod/github.com/olivere/elastic/v7@v7.0.22/bulk_processor.go:336 +0x1ce

[ppg-coder]