ES批量请求`es_rejected_execution_exception`

Question

我有一些 5M 条目的切片（为简单起见，假设每个条目都是一个字节切片，使用

getIndexerItem

函数映射到索引器项），我将其平均分配给 200 个 go 例程。然后每个 go 例程都会调用

push

函数，并使用长度为 5M/200 的切片。

根据我对

Refresh: wait_for

的理解，每当向弹性发出请求时，只有当该请求所做的更改对搜索可见时才会完成（IMO 将其转换为不再具有此特定请求的批量请求队列）。那为什么我会收到这个错误？

error indexing item: es_rejected_execution_exception:
rejected execution of processing of [358323543][indices:data/write/bulk[s][p]]: 
request: BulkShardRequest [[ankit-test][3]] containing [3424] requests blocking until refresh,
target allocation id: someId, primary term: 1 on EsThreadPoolExecutor
[
    name = machine_name/write, queue capacity = 200, org.elasticsearch.common.util.concurrent.EsThreadPoolExecutor@1f483ca1
    [Running, pool size = 32, active threads = 32, queued tasks = 200, completed tasks = 44390708]
]

所有条目都将进入相同的索引，

ankit-test

。

func (e *esClient) getIndexerItem(index string, id string, body []byte) esutil.BulkIndexerItem {
    return esutil.BulkIndexerItem{
        Index:        index,
        DocumentID:   id,
        Body:         bytes.NewReader(body),
        Action:       "index",
        DocumentType: "logs",
        OnFailure: func(_ context.Context, item esutil.BulkIndexerItem, res esutil.BulkIndexerResponseItem, err error) {
            if err != nil {
                fmt.Printf("error indexing item: %s\n", err.Error())
            } else {
                fmt.Printf("error indexing item: %s: %s\n", res.Error.Type, res.Error.Reason)
            }
        },
    }
}

func (e *esClient) push(data []esutil.BulkIndexerItem) (*esutil.BulkIndexerStats, error) {
    indexer, err := esutil.NewBulkIndexer(esutil.BulkIndexerConfig{
        Client: e.client,
        Refresh: "wait_for",
        NumWorkers: 1,
        OnError: func(ctx context.Context, err error) {
            fmt.Printf("received onError %s\n", err.Error())
        },
    })
    if err != nil {
        return nil, fmt.Errorf("error creating bulk indexer: %s", err)
    }

    ctx := context.Background()
    for _, d := range data {
        if err := indexer.Add(ctx, d); err != nil {
            fmt.Printf("error adding data to indexer: %s\n", err)
        }
    }
    if err := indexer.Close(ctx); err != nil {
        fmt.Printf("error flushing and closing indexer: %s\n", err)
    }

    indexerStats := indexer.Stats()
    return &indexerStats, nil
}

假设没有其他进程以任何方式与索引交互。

Answer 1

使用多个 ES 文档，我能够找到上述问题的解决方案。下面的回答是基于我的理解。如果您发现可以改进/纠正的地方，请发表评论。

---

这是请求生命周期：

1. golang es 客户端将多个请求合并为一个，并以单个批量请求的形式发送到服务器。单个批量请求可以包含发往多个索引和分片的文档。
2. 当批量请求到达集群中的某个节点（也称为协调节点）时，它会被完整地放入批量队列中，并由批量线程池中的线程处理。
3. 协调节点根据文档需要路由到的分片来拆分批量请求。 每个批量子请求都会转发到保存相应主分片的数据节点。批量子请求在该节点的批量队列中排队。如果队列上没有可用空间，则通知协调节点批量子请求已被拒绝。
4. 一旦所有子请求完成或被拒绝，就会创建响应并将其返回给客户端。有可能甚至很可能批量请求中只有部分文档被拒绝。

我的问题是我使用

refresh = false

（默认）发送请求。相反，应该使用

refresh = wait_for

。为什么？刷新提供了 3 种模式：

1. false：不执行刷新相关操作。此请求所做的更改将在请求返回后的某个时刻可见。在收到响应时，请求不必已完成。请求可能仍在节点的队列中。
2. true：操作发生后立即刷新相关主分片和副本分片。确保在发回响应之前请求已完成。请求已从节点队列中删除。
3. wait_for：等待请求所做的更改在回复之前通过刷新可见。与 true 不同的是，这不会强制立即刷新，而是等待刷新发生。比

   refresh = true

   便宜（就服务器负载而言），但仍然确保请求在发送回响应之前已完成。请求已从节点队列中删除。

所有数据都被重定向到同一个节点，并且由于

refresh = false

，在现有请求从队列中清除之前就返回了响应，这导致了溢出。