afterBulk() метод

[Katja]

Здравствуйте, подскажите, пожалуйста, с таким вопросом. Я индексирую документы с помощью BulkProcessor. У меня такой user case:

"An application has critical data and might decide to stop passing requests to bulk processor until the Elasticsearch cluster is up again."

Я нашла, что можно добавить в afterBulk() метод request.waitForActiveShards(). Но я неуверенна, достаточно ли этого. Спасибо и мой метод:

@Override
public void afterBulk(long executionId,
BulkRequest request,
Throwable failure) {
logger.warn("There is a problem with the burk request", failure);
request.waitForActiveShards();
}

[dadoonet (David Pilato)]

I moved your question

[Igor_Motov]

@Katja

Я нашла, что можно добавить в afterBurk() метод request.waitForActiveShards().

Где вы это нашли?

[Katja]

В методах, которые можно вызывать на объекте request. Теперь понимаю, что это неправильно. Остановилась на следующем, что в методе afterBulk() добавляю request: bulkProcessor.add(request). А потом уже в своей программе жду долго, а потом закрываю bulkProcessor. Но как остановить добавление документов в bulkProcessor не знаю, если в предыдущем bulk возникает ошибка.

[Igor_Motov]

Ну так вопрос в том, от чего возникла эта ошибка. Если она произошла потому, что сервер перегружен, то BulkProcessor эти ошибки перехватывает и реагирует в соответствии с BackoffPolicy. Если индекс не доступен, то запрос на индексирование будет ждать пока праймари не появится или свалиться по таймауту. Во всех остальных, ошибка скорее всего возникла из-за того, что проблема с самой записью. То есть добавляет ее снова и снова не имеет смысла.

[Katja]

@Igor_Motov

То есть добавляет ее снова и снова не имеет смысла.

Спасибо. По-моему ошибка возникла из-за java.lang.OutOfMemoryError: Java heap space. Поэтому индекс был недоступен. Попробую дать es больше памяти. Я правильно поняла, что я в afterBurk() могу поймать ошибки. А "to stop passing requests to bulk processor until the Elasticsearch cluster is up again" не получится.

[Igor_Motov]

Теоретически, можно запрашивать state индекса время от времени и перестать добавлять новые записи в BulkProcessor до тех пор, пока индекс не позеленеет. Вопрос в том, где эти записи в это время держать? Если у вас есть очередь какая-нибудь перед elasticsearch, то можно держать из в это очереди, туда же можно добавить записи из afterBulk, если вы уверены, что имеет смысл их еще раз попытаться проиндексировать.

[Katja]

Спасибо.