[jira] [Commented] (FLINK-11046) ElasticSearch6Connector cause thread blocked when index failed with retry
[
https://issues.apache.org/jira/browse/FLINK-11046?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16945072#comment-16945072
]
Suxing Lee commented on FLINK-11046:
Another situation that can lead to deadlock is the bug of es6.x itself. See
[https://github.com/elastic/elasticsearch/issues/47599] for details.
We can workaround the deadlock by changing the BackoffPolicy to noBackoff.
So we'd better set the elastic connection properties as follows in
ElasticSearch6Connector.
{code:java}
bulk.flush.backoff.enable: false
{code}
> ElasticSearch6Connector cause thread blocked when index failed with retry
> -
>
> Key: FLINK-11046
> URL: https://issues.apache.org/jira/browse/FLINK-11046
> Project: Flink
> Issue Type: Bug
> Components: Connectors / ElasticSearch
>Affects Versions: 1.6.2
>Reporter: luoguohao
>Assignee: xueyu
>Priority: Major
> Labels: pull-request-available
> Fix For: 1.7.2, 1.8.0
>
> Time Spent: 10m
> Remaining Estimate: 0h
>
> When i'm using es6 sink to index into es, bulk process with some exception
> catched, and i trying to reindex the document with the call
> `indexer.add(action)` in the `ActionRequestFailureHandler.onFailure()`
> method, but things goes incorrect. The call thread stuck there, and with the
> thread dump, i saw the `bulkprocessor` object was locked by other thread.
> {code:java}
> public interface ActionRequestFailureHandler extends Serializable {
> void onFailure(ActionRequest action, Throwable failure, int restStatusCode,
> RequestIndexer indexer) throws Throwable;
> }
> {code}
> After i read the code implemented in the `indexer.add(action)`, i find that
> `synchronized` is needed on each add operation.
> {code:java}
> private synchronized void internalAdd(DocWriteRequest request, @Nullable
> Object payload) {
> ensureOpen();
> bulkRequest.add(request, payload);
> executeIfNeeded();
> }
> {code}
> And, at i also noticed that `bulkprocessor` object would also locked in the
> bulk process thread.
> the bulk process operation is in the following code:
> {code:java}
> public void execute(BulkRequest bulkRequest, long executionId) {
> Runnable toRelease = () -> {};
> boolean bulkRequestSetupSuccessful = false;
> try {
> listener.beforeBulk(executionId, bulkRequest);
> semaphore.acquire();
> toRelease = semaphore::release;
> CountDownLatch latch = new CountDownLatch(1);
> retry.withBackoff(consumer, bulkRequest, new
> ActionListener() {
> @Override
> public void onResponse(BulkResponse response) {
> try {
> listener.afterBulk(executionId, bulkRequest, response);
> } finally {
> semaphore.release();
> latch.countDown();
> }
> }
> @Override
> public void onFailure(Exception e) {
> try {
> listener.afterBulk(executionId, bulkRequest, e);
> } finally {
> semaphore.release();
> latch.countDown();
> }
> }
> }, Settings.EMPTY);
> bulkRequestSetupSuccessful = true;
>if (concurrentRequests == 0) {
>latch.await();
> }
> } catch (InterruptedException e) {
> Thread.currentThread().interrupt();
> logger.info(() -> new ParameterizedMessage("Bulk request {} has been
> cancelled.", executionId), e);
> listener.afterBulk(executionId, bulkRequest, e);
> } catch (Exception e) {
> logger.warn(() -> new ParameterizedMessage("Failed to execute bulk
> request {}.", executionId), e);
> listener.afterBulk(executionId, bulkRequest, e);
> } finally {
> if (bulkRequestSetupSuccessful == false) { // if we fail on
> client.bulk() release the semaphore
> toRelease.run();
> }
> }
> }
> {code}
> As the read line i marked above, i think, that's the reason why the retry
> operation thread was block, because the the bulk process thread never release
> the lock on `bulkprocessor`. and, i also trying to figure out why the field
> `concurrentRequests` was set to zero. And i saw the the initialize for
> bulkprocessor in class `ElasticsearchSinkBase`:
> {code:java}
> protected BulkProcessor buildBulkProcessor(BulkProcessor.Listener listener) {
> ...
> BulkProcessor.Builder bulkProcessorBuilder =
> callBridge.createBulkProcessorBuilder(client, listener);
> // This makes flush() blocking
> bulkProcessorBuilder.setConcurrentRequests(0);
>
> ...
> return bulkProcessorBuilder.build();
> }
> {code}
> this field value was set to zero explicitly. So,
[jira] [Commented] (FLINK-11046) ElasticSearch6Connector cause thread blocked when index failed with retry
[
https://issues.apache.org/jira/browse/FLINK-11046?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16750891#comment-16750891
]
xueyu commented on FLINK-11046:
---
It looks that this kind of IndexRequest could result in reponse failure...
{code:java}
Requests.indexRequest()
.index(" :ind: ind : ")
.type(" :type: type : ")
.id(element)
.source("{data>\n<+++> :dw : dw : ", "whatisyourname");
{code}
> ElasticSearch6Connector cause thread blocked when index failed with retry
> -
>
> Key: FLINK-11046
> URL: https://issues.apache.org/jira/browse/FLINK-11046
> Project: Flink
> Issue Type: Bug
> Components: ElasticSearch Connector
>Affects Versions: 1.6.2
>Reporter: luoguohao
>Assignee: xueyu
>Priority: Major
>
> When i'm using es6 sink to index into es, bulk process with some exception
> catched, and i trying to reindex the document with the call
> `indexer.add(action)` in the `ActionRequestFailureHandler.onFailure()`
> method, but things goes incorrect. The call thread stuck there, and with the
> thread dump, i saw the `bulkprocessor` object was locked by other thread.
> {code:java}
> public interface ActionRequestFailureHandler extends Serializable {
> void onFailure(ActionRequest action, Throwable failure, int restStatusCode,
> RequestIndexer indexer) throws Throwable;
> }
> {code}
> After i read the code implemented in the `indexer.add(action)`, i find that
> `synchronized` is needed on each add operation.
> {code:java}
> private synchronized void internalAdd(DocWriteRequest request, @Nullable
> Object payload) {
> ensureOpen();
> bulkRequest.add(request, payload);
> executeIfNeeded();
> }
> {code}
> And, at i also noticed that `bulkprocessor` object would also locked in the
> bulk process thread.
> the bulk process operation is in the following code:
> {code:java}
> public void execute(BulkRequest bulkRequest, long executionId) {
> Runnable toRelease = () -> {};
> boolean bulkRequestSetupSuccessful = false;
> try {
> listener.beforeBulk(executionId, bulkRequest);
> semaphore.acquire();
> toRelease = semaphore::release;
> CountDownLatch latch = new CountDownLatch(1);
> retry.withBackoff(consumer, bulkRequest, new
> ActionListener() {
> @Override
> public void onResponse(BulkResponse response) {
> try {
> listener.afterBulk(executionId, bulkRequest, response);
> } finally {
> semaphore.release();
> latch.countDown();
> }
> }
> @Override
> public void onFailure(Exception e) {
> try {
> listener.afterBulk(executionId, bulkRequest, e);
> } finally {
> semaphore.release();
> latch.countDown();
> }
> }
> }, Settings.EMPTY);
> bulkRequestSetupSuccessful = true;
>if (concurrentRequests == 0) {
>latch.await();
> }
> } catch (InterruptedException e) {
> Thread.currentThread().interrupt();
> logger.info(() -> new ParameterizedMessage("Bulk request {} has been
> cancelled.", executionId), e);
> listener.afterBulk(executionId, bulkRequest, e);
> } catch (Exception e) {
> logger.warn(() -> new ParameterizedMessage("Failed to execute bulk
> request {}.", executionId), e);
> listener.afterBulk(executionId, bulkRequest, e);
> } finally {
> if (bulkRequestSetupSuccessful == false) { // if we fail on
> client.bulk() release the semaphore
> toRelease.run();
> }
> }
> }
> {code}
> As the read line i marked above, i think, that's the reason why the retry
> operation thread was block, because the the bulk process thread never release
> the lock on `bulkprocessor`. and, i also trying to figure out why the field
> `concurrentRequests` was set to zero. And i saw the the initialize for
> bulkprocessor in class `ElasticsearchSinkBase`:
> {code:java}
> protected BulkProcessor buildBulkProcessor(BulkProcessor.Listener listener) {
> ...
> BulkProcessor.Builder bulkProcessorBuilder =
> callBridge.createBulkProcessorBuilder(client, listener);
> // This makes flush() blocking
> bulkProcessorBuilder.setConcurrentRequests(0);
>
> ...
> return bulkProcessorBuilder.build();
> }
> {code}
> this field value was set to zero explicitly. So, all things seems to make
> sense, but i still wonder why the retry operation is not in the same thread
> as the bulk process execution, after i read the code, `bulkAsync` method
> might be the last puzzle.
> {code:java}
> @Override
> public
[jira] [Commented] (FLINK-11046) ElasticSearch6Connector cause thread blocked when index failed with retry
[
https://issues.apache.org/jira/browse/FLINK-11046?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16750025#comment-16750025
]
xueyu commented on FLINK-11046:
---
Hi, [~dawidwys][~tzulitai], I don't know how to make elastic6 response has
failure and then trigger onFailure callback in e2e test
Elasticsearch6SinkExample.java right now...Do you have any suggestions about
this..? Thanks.
I tried this and was not sure whether it is right..
{code:java}
private static IndexRequest createIndexRequest(String element,
ParameterTool parameterTool) {
Map json = new HashMap<>();
json.put("data", element);
if (element.startsWith("message #15")) {
return Requests.indexRequest()
.index(null)
.type(null)
.id(null)
.source(element);
} else {
return Requests.indexRequest()
.index(parameterTool.getRequired("index"))
.type(parameterTool.getRequired("type"))
.id(element)
.source(json);
}
}
{code}
> ElasticSearch6Connector cause thread blocked when index failed with retry
> -
>
> Key: FLINK-11046
> URL: https://issues.apache.org/jira/browse/FLINK-11046
> Project: Flink
> Issue Type: Bug
> Components: ElasticSearch Connector
>Affects Versions: 1.6.2
>Reporter: luoguohao
>Assignee: xueyu
>Priority: Major
>
> When i'm using es6 sink to index into es, bulk process with some exception
> catched, and i trying to reindex the document with the call
> `indexer.add(action)` in the `ActionRequestFailureHandler.onFailure()`
> method, but things goes incorrect. The call thread stuck there, and with the
> thread dump, i saw the `bulkprocessor` object was locked by other thread.
> {code:java}
> public interface ActionRequestFailureHandler extends Serializable {
> void onFailure(ActionRequest action, Throwable failure, int restStatusCode,
> RequestIndexer indexer) throws Throwable;
> }
> {code}
> After i read the code implemented in the `indexer.add(action)`, i find that
> `synchronized` is needed on each add operation.
> {code:java}
> private synchronized void internalAdd(DocWriteRequest request, @Nullable
> Object payload) {
> ensureOpen();
> bulkRequest.add(request, payload);
> executeIfNeeded();
> }
> {code}
> And, at i also noticed that `bulkprocessor` object would also locked in the
> bulk process thread.
> the bulk process operation is in the following code:
> {code:java}
> public void execute(BulkRequest bulkRequest, long executionId) {
> Runnable toRelease = () -> {};
> boolean bulkRequestSetupSuccessful = false;
> try {
> listener.beforeBulk(executionId, bulkRequest);
> semaphore.acquire();
> toRelease = semaphore::release;
> CountDownLatch latch = new CountDownLatch(1);
> retry.withBackoff(consumer, bulkRequest, new
> ActionListener() {
> @Override
> public void onResponse(BulkResponse response) {
> try {
> listener.afterBulk(executionId, bulkRequest, response);
> } finally {
> semaphore.release();
> latch.countDown();
> }
> }
> @Override
> public void onFailure(Exception e) {
> try {
> listener.afterBulk(executionId, bulkRequest, e);
> } finally {
> semaphore.release();
> latch.countDown();
> }
> }
> }, Settings.EMPTY);
> bulkRequestSetupSuccessful = true;
>if (concurrentRequests == 0) {
>latch.await();
> }
> } catch (InterruptedException e) {
> Thread.currentThread().interrupt();
> logger.info(() -> new ParameterizedMessage("Bulk request {} has been
> cancelled.", executionId), e);
> listener.afterBulk(executionId, bulkRequest, e);
> } catch (Exception e) {
> logger.warn(() -> new ParameterizedMessage("Failed to execute bulk
> request {}.", executionId), e);
> listener.afterBulk(executionId, bulkRequest, e);
> } finally {
> if (bulkRequestSetupSuccessful == false) { // if we fail on
> client.bulk() release the semaphore
> toRelease.run();
> }
> }
> }
> {code}
> As the read line i marked above, i think, that's the reason why the retry
> operation thread was block, because the the bulk
[jira] [Commented] (FLINK-11046) ElasticSearch6Connector cause thread blocked when index failed with retry
[
https://issues.apache.org/jira/browse/FLINK-11046?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16747831#comment-16747831
]
xueyu commented on FLINK-11046:
---
Hi, [~dawidwys], sorry that I only writed the codes and didn't write any test
and end-to-end test yet...The codes are on [my
branch|https://github.com/xueyumusic/flink/tree/es6-onfailure].
According to my current work status I estimated that still need about two weeks
from my side.. If this issue is urgent, please feel free to take it over.
Thank you and sorry for delay..., [~dawidwys]
> ElasticSearch6Connector cause thread blocked when index failed with retry
> -
>
> Key: FLINK-11046
> URL: https://issues.apache.org/jira/browse/FLINK-11046
> Project: Flink
> Issue Type: Bug
> Components: ElasticSearch Connector
>Affects Versions: 1.6.2
>Reporter: luoguohao
>Assignee: xueyu
>Priority: Major
>
> When i'm using es6 sink to index into es, bulk process with some exception
> catched, and i trying to reindex the document with the call
> `indexer.add(action)` in the `ActionRequestFailureHandler.onFailure()`
> method, but things goes incorrect. The call thread stuck there, and with the
> thread dump, i saw the `bulkprocessor` object was locked by other thread.
> {code:java}
> public interface ActionRequestFailureHandler extends Serializable {
> void onFailure(ActionRequest action, Throwable failure, int restStatusCode,
> RequestIndexer indexer) throws Throwable;
> }
> {code}
> After i read the code implemented in the `indexer.add(action)`, i find that
> `synchronized` is needed on each add operation.
> {code:java}
> private synchronized void internalAdd(DocWriteRequest request, @Nullable
> Object payload) {
> ensureOpen();
> bulkRequest.add(request, payload);
> executeIfNeeded();
> }
> {code}
> And, at i also noticed that `bulkprocessor` object would also locked in the
> bulk process thread.
> the bulk process operation is in the following code:
> {code:java}
> public void execute(BulkRequest bulkRequest, long executionId) {
> Runnable toRelease = () -> {};
> boolean bulkRequestSetupSuccessful = false;
> try {
> listener.beforeBulk(executionId, bulkRequest);
> semaphore.acquire();
> toRelease = semaphore::release;
> CountDownLatch latch = new CountDownLatch(1);
> retry.withBackoff(consumer, bulkRequest, new
> ActionListener() {
> @Override
> public void onResponse(BulkResponse response) {
> try {
> listener.afterBulk(executionId, bulkRequest, response);
> } finally {
> semaphore.release();
> latch.countDown();
> }
> }
> @Override
> public void onFailure(Exception e) {
> try {
> listener.afterBulk(executionId, bulkRequest, e);
> } finally {
> semaphore.release();
> latch.countDown();
> }
> }
> }, Settings.EMPTY);
> bulkRequestSetupSuccessful = true;
>if (concurrentRequests == 0) {
>latch.await();
> }
> } catch (InterruptedException e) {
> Thread.currentThread().interrupt();
> logger.info(() -> new ParameterizedMessage("Bulk request {} has been
> cancelled.", executionId), e);
> listener.afterBulk(executionId, bulkRequest, e);
> } catch (Exception e) {
> logger.warn(() -> new ParameterizedMessage("Failed to execute bulk
> request {}.", executionId), e);
> listener.afterBulk(executionId, bulkRequest, e);
> } finally {
> if (bulkRequestSetupSuccessful == false) { // if we fail on
> client.bulk() release the semaphore
> toRelease.run();
> }
> }
> }
> {code}
> As the read line i marked above, i think, that's the reason why the retry
> operation thread was block, because the the bulk process thread never release
> the lock on `bulkprocessor`. and, i also trying to figure out why the field
> `concurrentRequests` was set to zero. And i saw the the initialize for
> bulkprocessor in class `ElasticsearchSinkBase`:
> {code:java}
> protected BulkProcessor buildBulkProcessor(BulkProcessor.Listener listener) {
> ...
> BulkProcessor.Builder bulkProcessorBuilder =
> callBridge.createBulkProcessorBuilder(client, listener);
> // This makes flush() blocking
> bulkProcessorBuilder.setConcurrentRequests(0);
>
> ...
> return bulkProcessorBuilder.build();
> }
> {code}
> this field value was set to zero explicitly. So, all things seems to make
> sense, but i still wonder why the retry operation is not in the same thread
> as the bulk
[jira] [Commented] (FLINK-11046) ElasticSearch6Connector cause thread blocked when index failed with retry
[
https://issues.apache.org/jira/browse/FLINK-11046?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16747810#comment-16747810
]
Dawid Wysakowicz commented on FLINK-11046:
--
Hi [~xueyu] Did you manage to make any progress on this issue?
> ElasticSearch6Connector cause thread blocked when index failed with retry
> -
>
> Key: FLINK-11046
> URL: https://issues.apache.org/jira/browse/FLINK-11046
> Project: Flink
> Issue Type: Bug
> Components: ElasticSearch Connector
>Affects Versions: 1.6.2
>Reporter: luoguohao
>Assignee: xueyu
>Priority: Major
>
> When i'm using es6 sink to index into es, bulk process with some exception
> catched, and i trying to reindex the document with the call
> `indexer.add(action)` in the `ActionRequestFailureHandler.onFailure()`
> method, but things goes incorrect. The call thread stuck there, and with the
> thread dump, i saw the `bulkprocessor` object was locked by other thread.
> {code:java}
> public interface ActionRequestFailureHandler extends Serializable {
> void onFailure(ActionRequest action, Throwable failure, int restStatusCode,
> RequestIndexer indexer) throws Throwable;
> }
> {code}
> After i read the code implemented in the `indexer.add(action)`, i find that
> `synchronized` is needed on each add operation.
> {code:java}
> private synchronized void internalAdd(DocWriteRequest request, @Nullable
> Object payload) {
> ensureOpen();
> bulkRequest.add(request, payload);
> executeIfNeeded();
> }
> {code}
> And, at i also noticed that `bulkprocessor` object would also locked in the
> bulk process thread.
> the bulk process operation is in the following code:
> {code:java}
> public void execute(BulkRequest bulkRequest, long executionId) {
> Runnable toRelease = () -> {};
> boolean bulkRequestSetupSuccessful = false;
> try {
> listener.beforeBulk(executionId, bulkRequest);
> semaphore.acquire();
> toRelease = semaphore::release;
> CountDownLatch latch = new CountDownLatch(1);
> retry.withBackoff(consumer, bulkRequest, new
> ActionListener() {
> @Override
> public void onResponse(BulkResponse response) {
> try {
> listener.afterBulk(executionId, bulkRequest, response);
> } finally {
> semaphore.release();
> latch.countDown();
> }
> }
> @Override
> public void onFailure(Exception e) {
> try {
> listener.afterBulk(executionId, bulkRequest, e);
> } finally {
> semaphore.release();
> latch.countDown();
> }
> }
> }, Settings.EMPTY);
> bulkRequestSetupSuccessful = true;
>if (concurrentRequests == 0) {
>latch.await();
> }
> } catch (InterruptedException e) {
> Thread.currentThread().interrupt();
> logger.info(() -> new ParameterizedMessage("Bulk request {} has been
> cancelled.", executionId), e);
> listener.afterBulk(executionId, bulkRequest, e);
> } catch (Exception e) {
> logger.warn(() -> new ParameterizedMessage("Failed to execute bulk
> request {}.", executionId), e);
> listener.afterBulk(executionId, bulkRequest, e);
> } finally {
> if (bulkRequestSetupSuccessful == false) { // if we fail on
> client.bulk() release the semaphore
> toRelease.run();
> }
> }
> }
> {code}
> As the read line i marked above, i think, that's the reason why the retry
> operation thread was block, because the the bulk process thread never release
> the lock on `bulkprocessor`. and, i also trying to figure out why the field
> `concurrentRequests` was set to zero. And i saw the the initialize for
> bulkprocessor in class `ElasticsearchSinkBase`:
> {code:java}
> protected BulkProcessor buildBulkProcessor(BulkProcessor.Listener listener) {
> ...
> BulkProcessor.Builder bulkProcessorBuilder =
> callBridge.createBulkProcessorBuilder(client, listener);
> // This makes flush() blocking
> bulkProcessorBuilder.setConcurrentRequests(0);
>
> ...
> return bulkProcessorBuilder.build();
> }
> {code}
> this field value was set to zero explicitly. So, all things seems to make
> sense, but i still wonder why the retry operation is not in the same thread
> as the bulk process execution, after i read the code, `bulkAsync` method
> might be the last puzzle.
> {code:java}
> @Override
> public BulkProcessor.Builder createBulkProcessorBuilder(RestHighLevelClient
> client, BulkProcessor.Listener listener) {
> return BulkProcessor.builder(client::bulkAsync, listener);
> }
>
[jira] [Commented] (FLINK-11046) ElasticSearch6Connector cause thread blocked when index failed with retry
[
https://issues.apache.org/jira/browse/FLINK-11046?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16742981#comment-16742981
]
Tzu-Li (Gordon) Tai commented on FLINK-11046:
-
Just a side note for the implementation:
probably the Elasticsearch end-to-end test should be updated to cover this
scenario (re-indexing ES documents in the failure handler).
We should also make sure that the updated end-to-end test does actually fail
without the proposed fix.
> ElasticSearch6Connector cause thread blocked when index failed with retry
> -
>
> Key: FLINK-11046
> URL: https://issues.apache.org/jira/browse/FLINK-11046
> Project: Flink
> Issue Type: Bug
> Components: ElasticSearch Connector
>Affects Versions: 1.6.2
>Reporter: luoguohao
>Assignee: xueyu
>Priority: Major
>
> When i'm using es6 sink to index into es, bulk process with some exception
> catched, and i trying to reindex the document with the call
> `indexer.add(action)` in the `ActionRequestFailureHandler.onFailure()`
> method, but things goes incorrect. The call thread stuck there, and with the
> thread dump, i saw the `bulkprocessor` object was locked by other thread.
> {code:java}
> public interface ActionRequestFailureHandler extends Serializable {
> void onFailure(ActionRequest action, Throwable failure, int restStatusCode,
> RequestIndexer indexer) throws Throwable;
> }
> {code}
> After i read the code implemented in the `indexer.add(action)`, i find that
> `synchronized` is needed on each add operation.
> {code:java}
> private synchronized void internalAdd(DocWriteRequest request, @Nullable
> Object payload) {
> ensureOpen();
> bulkRequest.add(request, payload);
> executeIfNeeded();
> }
> {code}
> And, at i also noticed that `bulkprocessor` object would also locked in the
> bulk process thread.
> the bulk process operation is in the following code:
> {code:java}
> public void execute(BulkRequest bulkRequest, long executionId) {
> Runnable toRelease = () -> {};
> boolean bulkRequestSetupSuccessful = false;
> try {
> listener.beforeBulk(executionId, bulkRequest);
> semaphore.acquire();
> toRelease = semaphore::release;
> CountDownLatch latch = new CountDownLatch(1);
> retry.withBackoff(consumer, bulkRequest, new
> ActionListener() {
> @Override
> public void onResponse(BulkResponse response) {
> try {
> listener.afterBulk(executionId, bulkRequest, response);
> } finally {
> semaphore.release();
> latch.countDown();
> }
> }
> @Override
> public void onFailure(Exception e) {
> try {
> listener.afterBulk(executionId, bulkRequest, e);
> } finally {
> semaphore.release();
> latch.countDown();
> }
> }
> }, Settings.EMPTY);
> bulkRequestSetupSuccessful = true;
>if (concurrentRequests == 0) {
>latch.await();
> }
> } catch (InterruptedException e) {
> Thread.currentThread().interrupt();
> logger.info(() -> new ParameterizedMessage("Bulk request {} has been
> cancelled.", executionId), e);
> listener.afterBulk(executionId, bulkRequest, e);
> } catch (Exception e) {
> logger.warn(() -> new ParameterizedMessage("Failed to execute bulk
> request {}.", executionId), e);
> listener.afterBulk(executionId, bulkRequest, e);
> } finally {
> if (bulkRequestSetupSuccessful == false) { // if we fail on
> client.bulk() release the semaphore
> toRelease.run();
> }
> }
> }
> {code}
> As the read line i marked above, i think, that's the reason why the retry
> operation thread was block, because the the bulk process thread never release
> the lock on `bulkprocessor`. and, i also trying to figure out why the field
> `concurrentRequests` was set to zero. And i saw the the initialize for
> bulkprocessor in class `ElasticsearchSinkBase`:
> {code:java}
> protected BulkProcessor buildBulkProcessor(BulkProcessor.Listener listener) {
> ...
> BulkProcessor.Builder bulkProcessorBuilder =
> callBridge.createBulkProcessorBuilder(client, listener);
> // This makes flush() blocking
> bulkProcessorBuilder.setConcurrentRequests(0);
>
> ...
> return bulkProcessorBuilder.build();
> }
> {code}
> this field value was set to zero explicitly. So, all things seems to make
> sense, but i still wonder why the retry operation is not in the same thread
> as the bulk process execution, after i read the code, `bulkAsync` method
> might be the last puzzle.
[jira] [Commented] (FLINK-11046) ElasticSearch6Connector cause thread blocked when index failed with retry
[
https://issues.apache.org/jira/browse/FLINK-11046?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16742980#comment-16742980
]
Tzu-Li (Gordon) Tai commented on FLINK-11046:
-
[~xueyu] please feel free to ping me after opening the PR. Thanks for working
on this.
> ElasticSearch6Connector cause thread blocked when index failed with retry
> -
>
> Key: FLINK-11046
> URL: https://issues.apache.org/jira/browse/FLINK-11046
> Project: Flink
> Issue Type: Bug
> Components: ElasticSearch Connector
>Affects Versions: 1.6.2
>Reporter: luoguohao
>Assignee: xueyu
>Priority: Major
>
> When i'm using es6 sink to index into es, bulk process with some exception
> catched, and i trying to reindex the document with the call
> `indexer.add(action)` in the `ActionRequestFailureHandler.onFailure()`
> method, but things goes incorrect. The call thread stuck there, and with the
> thread dump, i saw the `bulkprocessor` object was locked by other thread.
> {code:java}
> public interface ActionRequestFailureHandler extends Serializable {
> void onFailure(ActionRequest action, Throwable failure, int restStatusCode,
> RequestIndexer indexer) throws Throwable;
> }
> {code}
> After i read the code implemented in the `indexer.add(action)`, i find that
> `synchronized` is needed on each add operation.
> {code:java}
> private synchronized void internalAdd(DocWriteRequest request, @Nullable
> Object payload) {
> ensureOpen();
> bulkRequest.add(request, payload);
> executeIfNeeded();
> }
> {code}
> And, at i also noticed that `bulkprocessor` object would also locked in the
> bulk process thread.
> the bulk process operation is in the following code:
> {code:java}
> public void execute(BulkRequest bulkRequest, long executionId) {
> Runnable toRelease = () -> {};
> boolean bulkRequestSetupSuccessful = false;
> try {
> listener.beforeBulk(executionId, bulkRequest);
> semaphore.acquire();
> toRelease = semaphore::release;
> CountDownLatch latch = new CountDownLatch(1);
> retry.withBackoff(consumer, bulkRequest, new
> ActionListener() {
> @Override
> public void onResponse(BulkResponse response) {
> try {
> listener.afterBulk(executionId, bulkRequest, response);
> } finally {
> semaphore.release();
> latch.countDown();
> }
> }
> @Override
> public void onFailure(Exception e) {
> try {
> listener.afterBulk(executionId, bulkRequest, e);
> } finally {
> semaphore.release();
> latch.countDown();
> }
> }
> }, Settings.EMPTY);
> bulkRequestSetupSuccessful = true;
>if (concurrentRequests == 0) {
>latch.await();
> }
> } catch (InterruptedException e) {
> Thread.currentThread().interrupt();
> logger.info(() -> new ParameterizedMessage("Bulk request {} has been
> cancelled.", executionId), e);
> listener.afterBulk(executionId, bulkRequest, e);
> } catch (Exception e) {
> logger.warn(() -> new ParameterizedMessage("Failed to execute bulk
> request {}.", executionId), e);
> listener.afterBulk(executionId, bulkRequest, e);
> } finally {
> if (bulkRequestSetupSuccessful == false) { // if we fail on
> client.bulk() release the semaphore
> toRelease.run();
> }
> }
> }
> {code}
> As the read line i marked above, i think, that's the reason why the retry
> operation thread was block, because the the bulk process thread never release
> the lock on `bulkprocessor`. and, i also trying to figure out why the field
> `concurrentRequests` was set to zero. And i saw the the initialize for
> bulkprocessor in class `ElasticsearchSinkBase`:
> {code:java}
> protected BulkProcessor buildBulkProcessor(BulkProcessor.Listener listener) {
> ...
> BulkProcessor.Builder bulkProcessorBuilder =
> callBridge.createBulkProcessorBuilder(client, listener);
> // This makes flush() blocking
> bulkProcessorBuilder.setConcurrentRequests(0);
>
> ...
> return bulkProcessorBuilder.build();
> }
> {code}
> this field value was set to zero explicitly. So, all things seems to make
> sense, but i still wonder why the retry operation is not in the same thread
> as the bulk process execution, after i read the code, `bulkAsync` method
> might be the last puzzle.
> {code:java}
> @Override
> public BulkProcessor.Builder createBulkProcessorBuilder(RestHighLevelClient
> client, BulkProcessor.Listener listener) {
> return
[jira] [Commented] (FLINK-11046) ElasticSearch6Connector cause thread blocked when index failed with retry
[
https://issues.apache.org/jira/browse/FLINK-11046?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16742665#comment-16742665
]
luoguohao commented on FLINK-11046:
---
[~tzulitai] yes, that's the point. thanks for giving a suggestion. i may not
have enough time working on it, maybe next time. currently, i just get around
this problem by re-indexing manually. [~xueyu] i am also looking forward to
your commit and apply the patch when it is available. Thx.
> ElasticSearch6Connector cause thread blocked when index failed with retry
> -
>
> Key: FLINK-11046
> URL: https://issues.apache.org/jira/browse/FLINK-11046
> Project: Flink
> Issue Type: Bug
> Components: ElasticSearch Connector
>Affects Versions: 1.6.2
>Reporter: luoguohao
>Assignee: xueyu
>Priority: Major
>
> When i'm using es6 sink to index into es, bulk process with some exception
> catched, and i trying to reindex the document with the call
> `indexer.add(action)` in the `ActionRequestFailureHandler.onFailure()`
> method, but things goes incorrect. The call thread stuck there, and with the
> thread dump, i saw the `bulkprocessor` object was locked by other thread.
> {code:java}
> public interface ActionRequestFailureHandler extends Serializable {
> void onFailure(ActionRequest action, Throwable failure, int restStatusCode,
> RequestIndexer indexer) throws Throwable;
> }
> {code}
> After i read the code implemented in the `indexer.add(action)`, i find that
> `synchronized` is needed on each add operation.
> {code:java}
> private synchronized void internalAdd(DocWriteRequest request, @Nullable
> Object payload) {
> ensureOpen();
> bulkRequest.add(request, payload);
> executeIfNeeded();
> }
> {code}
> And, at i also noticed that `bulkprocessor` object would also locked in the
> bulk process thread.
> the bulk process operation is in the following code:
> {code:java}
> public void execute(BulkRequest bulkRequest, long executionId) {
> Runnable toRelease = () -> {};
> boolean bulkRequestSetupSuccessful = false;
> try {
> listener.beforeBulk(executionId, bulkRequest);
> semaphore.acquire();
> toRelease = semaphore::release;
> CountDownLatch latch = new CountDownLatch(1);
> retry.withBackoff(consumer, bulkRequest, new
> ActionListener() {
> @Override
> public void onResponse(BulkResponse response) {
> try {
> listener.afterBulk(executionId, bulkRequest, response);
> } finally {
> semaphore.release();
> latch.countDown();
> }
> }
> @Override
> public void onFailure(Exception e) {
> try {
> listener.afterBulk(executionId, bulkRequest, e);
> } finally {
> semaphore.release();
> latch.countDown();
> }
> }
> }, Settings.EMPTY);
> bulkRequestSetupSuccessful = true;
>if (concurrentRequests == 0) {
>latch.await();
> }
> } catch (InterruptedException e) {
> Thread.currentThread().interrupt();
> logger.info(() -> new ParameterizedMessage("Bulk request {} has been
> cancelled.", executionId), e);
> listener.afterBulk(executionId, bulkRequest, e);
> } catch (Exception e) {
> logger.warn(() -> new ParameterizedMessage("Failed to execute bulk
> request {}.", executionId), e);
> listener.afterBulk(executionId, bulkRequest, e);
> } finally {
> if (bulkRequestSetupSuccessful == false) { // if we fail on
> client.bulk() release the semaphore
> toRelease.run();
> }
> }
> }
> {code}
> As the read line i marked above, i think, that's the reason why the retry
> operation thread was block, because the the bulk process thread never release
> the lock on `bulkprocessor`. and, i also trying to figure out why the field
> `concurrentRequests` was set to zero. And i saw the the initialize for
> bulkprocessor in class `ElasticsearchSinkBase`:
> {code:java}
> protected BulkProcessor buildBulkProcessor(BulkProcessor.Listener listener) {
> ...
> BulkProcessor.Builder bulkProcessorBuilder =
> callBridge.createBulkProcessorBuilder(client, listener);
> // This makes flush() blocking
> bulkProcessorBuilder.setConcurrentRequests(0);
>
> ...
> return bulkProcessorBuilder.build();
> }
> {code}
> this field value was set to zero explicitly. So, all things seems to make
> sense, but i still wonder why the retry operation is not in the same thread
> as the bulk process execution, after i read the code, `bulkAsync` method
> might be the last puzzle.
>
[jira] [Commented] (FLINK-11046) ElasticSearch6Connector cause thread blocked when index failed with retry
[
https://issues.apache.org/jira/browse/FLINK-11046?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16742183#comment-16742183
]
Dawid Wysakowicz commented on FLINK-11046:
--
Hi [~xueyu],
I just started, so I can leave it for you.
> ElasticSearch6Connector cause thread blocked when index failed with retry
> -
>
> Key: FLINK-11046
> URL: https://issues.apache.org/jira/browse/FLINK-11046
> Project: Flink
> Issue Type: Bug
> Components: ElasticSearch Connector
>Affects Versions: 1.6.2
>Reporter: luoguohao
>Assignee: Dawid Wysakowicz
>Priority: Major
>
> When i'm using es6 sink to index into es, bulk process with some exception
> catched, and i trying to reindex the document with the call
> `indexer.add(action)` in the `ActionRequestFailureHandler.onFailure()`
> method, but things goes incorrect. The call thread stuck there, and with the
> thread dump, i saw the `bulkprocessor` object was locked by other thread.
> {code:java}
> public interface ActionRequestFailureHandler extends Serializable {
> void onFailure(ActionRequest action, Throwable failure, int restStatusCode,
> RequestIndexer indexer) throws Throwable;
> }
> {code}
> After i read the code implemented in the `indexer.add(action)`, i find that
> `synchronized` is needed on each add operation.
> {code:java}
> private synchronized void internalAdd(DocWriteRequest request, @Nullable
> Object payload) {
> ensureOpen();
> bulkRequest.add(request, payload);
> executeIfNeeded();
> }
> {code}
> And, at i also noticed that `bulkprocessor` object would also locked in the
> bulk process thread.
> the bulk process operation is in the following code:
> {code:java}
> public void execute(BulkRequest bulkRequest, long executionId) {
> Runnable toRelease = () -> {};
> boolean bulkRequestSetupSuccessful = false;
> try {
> listener.beforeBulk(executionId, bulkRequest);
> semaphore.acquire();
> toRelease = semaphore::release;
> CountDownLatch latch = new CountDownLatch(1);
> retry.withBackoff(consumer, bulkRequest, new
> ActionListener() {
> @Override
> public void onResponse(BulkResponse response) {
> try {
> listener.afterBulk(executionId, bulkRequest, response);
> } finally {
> semaphore.release();
> latch.countDown();
> }
> }
> @Override
> public void onFailure(Exception e) {
> try {
> listener.afterBulk(executionId, bulkRequest, e);
> } finally {
> semaphore.release();
> latch.countDown();
> }
> }
> }, Settings.EMPTY);
> bulkRequestSetupSuccessful = true;
>if (concurrentRequests == 0) {
>latch.await();
> }
> } catch (InterruptedException e) {
> Thread.currentThread().interrupt();
> logger.info(() -> new ParameterizedMessage("Bulk request {} has been
> cancelled.", executionId), e);
> listener.afterBulk(executionId, bulkRequest, e);
> } catch (Exception e) {
> logger.warn(() -> new ParameterizedMessage("Failed to execute bulk
> request {}.", executionId), e);
> listener.afterBulk(executionId, bulkRequest, e);
> } finally {
> if (bulkRequestSetupSuccessful == false) { // if we fail on
> client.bulk() release the semaphore
> toRelease.run();
> }
> }
> }
> {code}
> As the read line i marked above, i think, that's the reason why the retry
> operation thread was block, because the the bulk process thread never release
> the lock on `bulkprocessor`. and, i also trying to figure out why the field
> `concurrentRequests` was set to zero. And i saw the the initialize for
> bulkprocessor in class `ElasticsearchSinkBase`:
> {code:java}
> protected BulkProcessor buildBulkProcessor(BulkProcessor.Listener listener) {
> ...
> BulkProcessor.Builder bulkProcessorBuilder =
> callBridge.createBulkProcessorBuilder(client, listener);
> // This makes flush() blocking
> bulkProcessorBuilder.setConcurrentRequests(0);
>
> ...
> return bulkProcessorBuilder.build();
> }
> {code}
> this field value was set to zero explicitly. So, all things seems to make
> sense, but i still wonder why the retry operation is not in the same thread
> as the bulk process execution, after i read the code, `bulkAsync` method
> might be the last puzzle.
> {code:java}
> @Override
> public BulkProcessor.Builder createBulkProcessorBuilder(RestHighLevelClient
> client, BulkProcessor.Listener listener) {
> return BulkProcessor.builder(client::bulkAsync, listener);
> }
>
[jira] [Commented] (FLINK-11046) ElasticSearch6Connector cause thread blocked when index failed with retry
[
https://issues.apache.org/jira/browse/FLINK-11046?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16742179#comment-16742179
]
xueyu commented on FLINK-11046:
---
Hi, [~dawidwys], have you worked on this issue? If not could you please assign
this issue to me..I investigated this issue these two days and have some ideas
to fix it and wanted to have a try on it. According to [~tzulitai] comments, my
thought was writing a new RequestIndexer which uses BulkRequest to buffer
action requests. I was a little late to assign it... Thanks~
> ElasticSearch6Connector cause thread blocked when index failed with retry
> -
>
> Key: FLINK-11046
> URL: https://issues.apache.org/jira/browse/FLINK-11046
> Project: Flink
> Issue Type: Bug
> Components: ElasticSearch Connector
>Affects Versions: 1.6.2
>Reporter: luoguohao
>Assignee: Dawid Wysakowicz
>Priority: Major
>
> When i'm using es6 sink to index into es, bulk process with some exception
> catched, and i trying to reindex the document with the call
> `indexer.add(action)` in the `ActionRequestFailureHandler.onFailure()`
> method, but things goes incorrect. The call thread stuck there, and with the
> thread dump, i saw the `bulkprocessor` object was locked by other thread.
> {code:java}
> public interface ActionRequestFailureHandler extends Serializable {
> void onFailure(ActionRequest action, Throwable failure, int restStatusCode,
> RequestIndexer indexer) throws Throwable;
> }
> {code}
> After i read the code implemented in the `indexer.add(action)`, i find that
> `synchronized` is needed on each add operation.
> {code:java}
> private synchronized void internalAdd(DocWriteRequest request, @Nullable
> Object payload) {
> ensureOpen();
> bulkRequest.add(request, payload);
> executeIfNeeded();
> }
> {code}
> And, at i also noticed that `bulkprocessor` object would also locked in the
> bulk process thread.
> the bulk process operation is in the following code:
> {code:java}
> public void execute(BulkRequest bulkRequest, long executionId) {
> Runnable toRelease = () -> {};
> boolean bulkRequestSetupSuccessful = false;
> try {
> listener.beforeBulk(executionId, bulkRequest);
> semaphore.acquire();
> toRelease = semaphore::release;
> CountDownLatch latch = new CountDownLatch(1);
> retry.withBackoff(consumer, bulkRequest, new
> ActionListener() {
> @Override
> public void onResponse(BulkResponse response) {
> try {
> listener.afterBulk(executionId, bulkRequest, response);
> } finally {
> semaphore.release();
> latch.countDown();
> }
> }
> @Override
> public void onFailure(Exception e) {
> try {
> listener.afterBulk(executionId, bulkRequest, e);
> } finally {
> semaphore.release();
> latch.countDown();
> }
> }
> }, Settings.EMPTY);
> bulkRequestSetupSuccessful = true;
>if (concurrentRequests == 0) {
>latch.await();
> }
> } catch (InterruptedException e) {
> Thread.currentThread().interrupt();
> logger.info(() -> new ParameterizedMessage("Bulk request {} has been
> cancelled.", executionId), e);
> listener.afterBulk(executionId, bulkRequest, e);
> } catch (Exception e) {
> logger.warn(() -> new ParameterizedMessage("Failed to execute bulk
> request {}.", executionId), e);
> listener.afterBulk(executionId, bulkRequest, e);
> } finally {
> if (bulkRequestSetupSuccessful == false) { // if we fail on
> client.bulk() release the semaphore
> toRelease.run();
> }
> }
> }
> {code}
> As the read line i marked above, i think, that's the reason why the retry
> operation thread was block, because the the bulk process thread never release
> the lock on `bulkprocessor`. and, i also trying to figure out why the field
> `concurrentRequests` was set to zero. And i saw the the initialize for
> bulkprocessor in class `ElasticsearchSinkBase`:
> {code:java}
> protected BulkProcessor buildBulkProcessor(BulkProcessor.Listener listener) {
> ...
> BulkProcessor.Builder bulkProcessorBuilder =
> callBridge.createBulkProcessorBuilder(client, listener);
> // This makes flush() blocking
> bulkProcessorBuilder.setConcurrentRequests(0);
>
> ...
> return bulkProcessorBuilder.build();
> }
> {code}
> this field value was set to zero explicitly. So, all things seems to make
> sense, but i still wonder why the retry operation is not in the same thread
> as the bulk process
[jira] [Commented] (FLINK-11046) ElasticSearch6Connector cause thread blocked when index failed with retry
[
https://issues.apache.org/jira/browse/FLINK-11046?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16721348#comment-16721348
]
Tzu-Li (Gordon) Tai commented on FLINK-11046:
-
I propose the following fix:
The {{RequestIndexer}} instance provided to the {{ActionRequestFailureHandler}}
shouldn't be the same instance as the one used for indexing incoming records
(i.e. the one used in {{invoke}} method of the sink).
Instead, it should be a separate instance, which buffers any requests that the
user attempts to re-index in the failure handler.
In {{invoke}}, before processing the next element, we always check if there are
buffered requests from the failure handler that needs to be added to the actual
request indexer.
[~luoguohao] you mentioned that you would want to try fixing this. Do you want
to take a try on this?
> ElasticSearch6Connector cause thread blocked when index failed with retry
> -
>
> Key: FLINK-11046
> URL: https://issues.apache.org/jira/browse/FLINK-11046
> Project: Flink
> Issue Type: Bug
> Components: ElasticSearch Connector
>Affects Versions: 1.6.2
>Reporter: luoguohao
>Priority: Major
>
> When i'm using es6 sink to index into es, bulk process with some exception
> catched, and i trying to reindex the document with the call
> `indexer.add(action)` in the `ActionRequestFailureHandler.onFailure()`
> method, but things goes incorrect. The call thread stuck there, and with the
> thread dump, i saw the `bulkprocessor` object was locked by other thread.
> {code:java}
> public interface ActionRequestFailureHandler extends Serializable {
> void onFailure(ActionRequest action, Throwable failure, int restStatusCode,
> RequestIndexer indexer) throws Throwable;
> }
> {code}
> After i read the code implemented in the `indexer.add(action)`, i find that
> `synchronized` is needed on each add operation.
> {code:java}
> private synchronized void internalAdd(DocWriteRequest request, @Nullable
> Object payload) {
> ensureOpen();
> bulkRequest.add(request, payload);
> executeIfNeeded();
> }
> {code}
> And, at i also noticed that `bulkprocessor` object would also locked in the
> bulk process thread.
> the bulk process operation is in the following code:
> {code:java}
> public void execute(BulkRequest bulkRequest, long executionId) {
> Runnable toRelease = () -> {};
> boolean bulkRequestSetupSuccessful = false;
> try {
> listener.beforeBulk(executionId, bulkRequest);
> semaphore.acquire();
> toRelease = semaphore::release;
> CountDownLatch latch = new CountDownLatch(1);
> retry.withBackoff(consumer, bulkRequest, new
> ActionListener() {
> @Override
> public void onResponse(BulkResponse response) {
> try {
> listener.afterBulk(executionId, bulkRequest, response);
> } finally {
> semaphore.release();
> latch.countDown();
> }
> }
> @Override
> public void onFailure(Exception e) {
> try {
> listener.afterBulk(executionId, bulkRequest, e);
> } finally {
> semaphore.release();
> latch.countDown();
> }
> }
> }, Settings.EMPTY);
> bulkRequestSetupSuccessful = true;
>if (concurrentRequests == 0) {
>latch.await();
> }
> } catch (InterruptedException e) {
> Thread.currentThread().interrupt();
> logger.info(() -> new ParameterizedMessage("Bulk request {} has been
> cancelled.", executionId), e);
> listener.afterBulk(executionId, bulkRequest, e);
> } catch (Exception e) {
> logger.warn(() -> new ParameterizedMessage("Failed to execute bulk
> request {}.", executionId), e);
> listener.afterBulk(executionId, bulkRequest, e);
> } finally {
> if (bulkRequestSetupSuccessful == false) { // if we fail on
> client.bulk() release the semaphore
> toRelease.run();
> }
> }
> }
> {code}
> As the read line i marked above, i think, that's the reason why the retry
> operation thread was block, because the the bulk process thread never release
> the lock on `bulkprocessor`. and, i also trying to figure out why the field
> `concurrentRequests` was set to zero. And i saw the the initialize for
> bulkprocessor in class `ElasticsearchSinkBase`:
> {code:java}
> protected BulkProcessor buildBulkProcessor(BulkProcessor.Listener listener) {
> ...
> BulkProcessor.Builder bulkProcessorBuilder =
> callBridge.createBulkProcessorBuilder(client, listener);
> // This makes flush() blocking
>
[jira] [Commented] (FLINK-11046) ElasticSearch6Connector cause thread blocked when index failed with retry
[
https://issues.apache.org/jira/browse/FLINK-11046?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16721314#comment-16721314
]
Tzu-Li (Gordon) Tai commented on FLINK-11046:
-
Ah , I think I understand the deadlock now.
It's because the we're calling the user provided
{{ActionRequestFailureHandler.onFailure(...)}} inside the {{afterBulk}}
callback. The lock is only released when the {{afterBulk}} method returns.
So, the deadlock is:
1. The {{BulkProcessor}} flushes, and one of the document indexing failed,
which invokes the user's {{ActionRequestFailureHandler.onFailure(...)}}. At
this point, the lock on {{BulkProcessor}} isn't released yet, because the
{{onFailure}} call is part of the bulk processor's flush callback.
2. Within {{ActionRequestFailureHandler.onFailure(...)}}, in your case you
added some new documents to be indexed. Upon adding, the {{BulkProcessor}}
would try to flush again, but the lock wasn't released yet and therefore
deadlock.
So, the re-indexing thread (i.e. the async callback) should have been blocked
on:
[https://github.com/elastic/elasticsearch/blob/v6.3.1/server/src/main/java/org/elasticsearch/action/bulk/BulkRequestHandler.java#L60]
While the main task thread should have been blocked on:
[https://github.com/elastic/elasticsearch/blob/v6.3.1/server/src/main/java/org/elasticsearch/action/bulk/BulkRequestHandler.java#L86]
Could you confirm this and see if the analysis makes sense to you?
[~luoguohao]
> ElasticSearch6Connector cause thread blocked when index failed with retry
> -
>
> Key: FLINK-11046
> URL: https://issues.apache.org/jira/browse/FLINK-11046
> Project: Flink
> Issue Type: Bug
> Components: ElasticSearch Connector
>Affects Versions: 1.6.2
>Reporter: luoguohao
>Priority: Major
>
> When i'm using es6 sink to index into es, bulk process with some exception
> catched, and i trying to reindex the document with the call
> `indexer.add(action)` in the `ActionRequestFailureHandler.onFailure()`
> method, but things goes incorrect. The call thread stuck there, and with the
> thread dump, i saw the `bulkprocessor` object was locked by other thread.
> {code:java}
> public interface ActionRequestFailureHandler extends Serializable {
> void onFailure(ActionRequest action, Throwable failure, int restStatusCode,
> RequestIndexer indexer) throws Throwable;
> }
> {code}
> After i read the code implemented in the `indexer.add(action)`, i find that
> `synchronized` is needed on each add operation.
> {code:java}
> private synchronized void internalAdd(DocWriteRequest request, @Nullable
> Object payload) {
> ensureOpen();
> bulkRequest.add(request, payload);
> executeIfNeeded();
> }
> {code}
> And, at i also noticed that `bulkprocessor` object would also locked in the
> bulk process thread.
> the bulk process operation is in the following code:
> {code:java}
> public void execute(BulkRequest bulkRequest, long executionId) {
> Runnable toRelease = () -> {};
> boolean bulkRequestSetupSuccessful = false;
> try {
> listener.beforeBulk(executionId, bulkRequest);
> semaphore.acquire();
> toRelease = semaphore::release;
> CountDownLatch latch = new CountDownLatch(1);
> retry.withBackoff(consumer, bulkRequest, new
> ActionListener() {
> @Override
> public void onResponse(BulkResponse response) {
> try {
> listener.afterBulk(executionId, bulkRequest, response);
> } finally {
> semaphore.release();
> latch.countDown();
> }
> }
> @Override
> public void onFailure(Exception e) {
> try {
> listener.afterBulk(executionId, bulkRequest, e);
> } finally {
> semaphore.release();
> latch.countDown();
> }
> }
> }, Settings.EMPTY);
> bulkRequestSetupSuccessful = true;
>if (concurrentRequests == 0) {
>latch.await();
> }
> } catch (InterruptedException e) {
> Thread.currentThread().interrupt();
> logger.info(() -> new ParameterizedMessage("Bulk request {} has been
> cancelled.", executionId), e);
> listener.afterBulk(executionId, bulkRequest, e);
> } catch (Exception e) {
> logger.warn(() -> new ParameterizedMessage("Failed to execute bulk
> request {}.", executionId), e);
> listener.afterBulk(executionId, bulkRequest, e);
> } finally {
> if (bulkRequestSetupSuccessful == false) { // if we fail on
> client.bulk() release the semaphore
> toRelease.run();
> }
> }
> }
> {code}
[jira] [Commented] (FLINK-11046) ElasticSearch6Connector cause thread blocked when index failed with retry
[
https://issues.apache.org/jira/browse/FLINK-11046?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16721296#comment-16721296
]
Tzu-Li (Gordon) Tai commented on FLINK-11046:
-
This seems a bit odd.
While concurrent requests is indeed set to 0 and therefore only a single bulk
request will be allowed to be executed and new index accumulations are blocked
during the process, the lock should have been released after the bulk request
finishes and un-block the new index addition.
After all, the {{BulkProcessor}} is supposed to be thread-safe:
[http://javadoc.kyubu.de/elasticsearch/HEAD/org/elasticsearch/action/bulk/BulkProcessor.html.]
> ElasticSearch6Connector cause thread blocked when index failed with retry
> -
>
> Key: FLINK-11046
> URL: https://issues.apache.org/jira/browse/FLINK-11046
> Project: Flink
> Issue Type: Bug
> Components: ElasticSearch Connector
>Affects Versions: 1.6.2
>Reporter: luoguohao
>Priority: Major
>
> When i'm using es6 sink to index into es, bulk process with some exception
> catched, and i trying to reindex the document with the call
> `indexer.add(action)` in the `ActionRequestFailureHandler.onFailure()`
> method, but things goes incorrect. The call thread stuck there, and with the
> thread dump, i saw the `bulkprocessor` object was locked by other thread.
> {code:java}
> public interface ActionRequestFailureHandler extends Serializable {
> void onFailure(ActionRequest action, Throwable failure, int restStatusCode,
> RequestIndexer indexer) throws Throwable;
> }
> {code}
> After i read the code implemented in the `indexer.add(action)`, i find that
> `synchronized` is needed on each add operation.
> {code:java}
> private synchronized void internalAdd(DocWriteRequest request, @Nullable
> Object payload) {
> ensureOpen();
> bulkRequest.add(request, payload);
> executeIfNeeded();
> }
> {code}
> And, at i also noticed that `bulkprocessor` object would also locked in the
> bulk process thread.
> the bulk process operation is in the following code:
> {code:java}
> public void execute(BulkRequest bulkRequest, long executionId) {
> Runnable toRelease = () -> {};
> boolean bulkRequestSetupSuccessful = false;
> try {
> listener.beforeBulk(executionId, bulkRequest);
> semaphore.acquire();
> toRelease = semaphore::release;
> CountDownLatch latch = new CountDownLatch(1);
> retry.withBackoff(consumer, bulkRequest, new
> ActionListener() {
> @Override
> public void onResponse(BulkResponse response) {
> try {
> listener.afterBulk(executionId, bulkRequest, response);
> } finally {
> semaphore.release();
> latch.countDown();
> }
> }
> @Override
> public void onFailure(Exception e) {
> try {
> listener.afterBulk(executionId, bulkRequest, e);
> } finally {
> semaphore.release();
> latch.countDown();
> }
> }
> }, Settings.EMPTY);
> bulkRequestSetupSuccessful = true;
>if (concurrentRequests == 0) {
>latch.await();
> }
> } catch (InterruptedException e) {
> Thread.currentThread().interrupt();
> logger.info(() -> new ParameterizedMessage("Bulk request {} has been
> cancelled.", executionId), e);
> listener.afterBulk(executionId, bulkRequest, e);
> } catch (Exception e) {
> logger.warn(() -> new ParameterizedMessage("Failed to execute bulk
> request {}.", executionId), e);
> listener.afterBulk(executionId, bulkRequest, e);
> } finally {
> if (bulkRequestSetupSuccessful == false) { // if we fail on
> client.bulk() release the semaphore
> toRelease.run();
> }
> }
> }
> {code}
> As the read line i marked above, i think, that's the reason why the retry
> operation thread was block, because the the bulk process thread never release
> the lock on `bulkprocessor`. and, i also trying to figure out why the field
> `concurrentRequests` was set to zero. And i saw the the initialize for
> bulkprocessor in class `ElasticsearchSinkBase`:
> {code:java}
> protected BulkProcessor buildBulkProcessor(BulkProcessor.Listener listener) {
> ...
> BulkProcessor.Builder bulkProcessorBuilder =
> callBridge.createBulkProcessorBuilder(client, listener);
> // This makes flush() blocking
> bulkProcessorBuilder.setConcurrentRequests(0);
>
> ...
> return bulkProcessorBuilder.build();
> }
> {code}
> this field value was set to zero explicitly. So, all things seems to make
> sense, but i still
[jira] [Commented] (FLINK-11046) ElasticSearch6Connector cause thread blocked when index failed with retry
[
https://issues.apache.org/jira/browse/FLINK-11046?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16708469#comment-16708469
]
luoguohao commented on FLINK-11046:
---
HI, [~tzulitai],
Thank you for your reply. Because our elasticSearch scheme is configured as
strict mode, so if some unexpected `strict_dynamic_mapping_exception` be
thrown, i would check if the field is invalid, otherwise, remapping the schema,
and re-adding the index operation into the `RequestIndexer` instance.
The snippet of the code would as simple as the follow:
{code:java}
class LoadTraitIntoEsFailureHandler
extends ActionRequestFailureHandler {
override def onFailure(action: ActionRequest,
failure: Throwable,
restStatusCode: Int,
indexer: RequestIndexer): Unit = {
failure match {
case exception: ElasticsearchException
if isStrictDynamicMappingException(exception) ⇒
val updateAction = action.asInstanceOf[UpdateRequest]
// remapping the es schema if needed
...
// finally, re-adding the action if needed
indexer.add(updateAction)
case others ⇒
// do nothing
}
}
}{code}
That's all things for the retry operation.
> ElasticSearch6Connector cause thread blocked when index failed with retry
> -
>
> Key: FLINK-11046
> URL: https://issues.apache.org/jira/browse/FLINK-11046
> Project: Flink
> Issue Type: Bug
> Components: ElasticSearch Connector
>Affects Versions: 1.6.2
>Reporter: luoguohao
>Priority: Major
>
> When i'm using es6 sink to index into es, bulk process with some exception
> catched, and i trying to reindex the document with the call
> `indexer.add(action)` in the `ActionRequestFailureHandler.onFailure()`
> method, but things goes incorrect. The call thread stuck there, and with the
> thread dump, i saw the `bulkprocessor` object was locked by other thread.
> {code:java}
> public interface ActionRequestFailureHandler extends Serializable {
> void onFailure(ActionRequest action, Throwable failure, int restStatusCode,
> RequestIndexer indexer) throws Throwable;
> }
> {code}
> After i read the code implemented in the `indexer.add(action)`, i find that
> `synchronized` is needed on each add operation.
> {code:java}
> private synchronized void internalAdd(DocWriteRequest request, @Nullable
> Object payload) {
> ensureOpen();
> bulkRequest.add(request, payload);
> executeIfNeeded();
> }
> {code}
> And, at i also noticed that `bulkprocessor` object would also locked in the
> bulk process thread.
> the bulk process operation is in the following code:
> {code:java}
> public void execute(BulkRequest bulkRequest, long executionId) {
> Runnable toRelease = () -> {};
> boolean bulkRequestSetupSuccessful = false;
> try {
> listener.beforeBulk(executionId, bulkRequest);
> semaphore.acquire();
> toRelease = semaphore::release;
> CountDownLatch latch = new CountDownLatch(1);
> retry.withBackoff(consumer, bulkRequest, new
> ActionListener() {
> @Override
> public void onResponse(BulkResponse response) {
> try {
> listener.afterBulk(executionId, bulkRequest, response);
> } finally {
> semaphore.release();
> latch.countDown();
> }
> }
> @Override
> public void onFailure(Exception e) {
> try {
> listener.afterBulk(executionId, bulkRequest, e);
> } finally {
> semaphore.release();
> latch.countDown();
> }
> }
> }, Settings.EMPTY);
> bulkRequestSetupSuccessful = true;
>if (concurrentRequests == 0) {
>latch.await();
> }
> } catch (InterruptedException e) {
> Thread.currentThread().interrupt();
> logger.info(() -> new ParameterizedMessage("Bulk request {} has been
> cancelled.", executionId), e);
> listener.afterBulk(executionId, bulkRequest, e);
> } catch (Exception e) {
> logger.warn(() -> new ParameterizedMessage("Failed to execute bulk
> request {}.", executionId), e);
> listener.afterBulk(executionId, bulkRequest, e);
> } finally {
> if (bulkRequestSetupSuccessful == false) { // if we fail on
> client.bulk() release the semaphore
> toRelease.run();
> }
> }
> }
> {code}
> As the read line i marked above, i think, that's the reason why the retry
> operation thread was block, because the the bulk process thread never release
> the lock on `bulkprocessor`. and, i also trying to figure out why
[jira] [Commented] (FLINK-11046) ElasticSearch6Connector cause thread blocked when index failed with retry
[
https://issues.apache.org/jira/browse/FLINK-11046?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16708337#comment-16708337
]
Tzu-Li (Gordon) Tai commented on FLINK-11046:
-
Hi [~luoguohao],
So, to clarify things a bit:
In your Elasticsearch function's failure handler, you're catching an exception
and readding documents to the indexer?
What exactly is the exception that you are getting?
If possible, could you also give me a more complete snippet of the code that
you're encountering this problem? That would help with figuring out the problem
on my side.
> ElasticSearch6Connector cause thread blocked when index failed with retry
> -
>
> Key: FLINK-11046
> URL: https://issues.apache.org/jira/browse/FLINK-11046
> Project: Flink
> Issue Type: Bug
> Components: ElasticSearch Connector
>Affects Versions: 1.6.2
>Reporter: luoguohao
>Priority: Major
>
> When i'm using es6 sink to index into es, bulk process with some exception
> catched, and i trying to reindex the document with the call
> `indexer.add(action)` in the `ActionRequestFailureHandler.onFailure()`
> method, but things goes incorrect. The call thread stuck there, and with the
> thread dump, i saw the `bulkprocessor` object was locked by other thread.
> {code:java}
> public interface ActionRequestFailureHandler extends Serializable {
> void onFailure(ActionRequest action, Throwable failure, int restStatusCode,
> RequestIndexer indexer) throws Throwable;
> }
> {code}
> After i read the code implemented in the `indexer.add(action)`, i find that
> `synchronized` is needed on each add operation.
> {code:java}
> private synchronized void internalAdd(DocWriteRequest request, @Nullable
> Object payload) {
> ensureOpen();
> bulkRequest.add(request, payload);
> executeIfNeeded();
> }
> {code}
> And, at i also noticed that `bulkprocessor` object would also locked in the
> bulk process thread.
> the bulk process operation is in the following code:
> {code:java}
> public void execute(BulkRequest bulkRequest, long executionId) {
> Runnable toRelease = () -> {};
> boolean bulkRequestSetupSuccessful = false;
> try {
> listener.beforeBulk(executionId, bulkRequest);
> semaphore.acquire();
> toRelease = semaphore::release;
> CountDownLatch latch = new CountDownLatch(1);
> retry.withBackoff(consumer, bulkRequest, new
> ActionListener() {
> @Override
> public void onResponse(BulkResponse response) {
> try {
> listener.afterBulk(executionId, bulkRequest, response);
> } finally {
> semaphore.release();
> latch.countDown();
> }
> }
> @Override
> public void onFailure(Exception e) {
> try {
> listener.afterBulk(executionId, bulkRequest, e);
> } finally {
> semaphore.release();
> latch.countDown();
> }
> }
> }, Settings.EMPTY);
> bulkRequestSetupSuccessful = true;
>if (concurrentRequests == 0) {
>latch.await();
> }
> } catch (InterruptedException e) {
> Thread.currentThread().interrupt();
> logger.info(() -> new ParameterizedMessage("Bulk request {} has been
> cancelled.", executionId), e);
> listener.afterBulk(executionId, bulkRequest, e);
> } catch (Exception e) {
> logger.warn(() -> new ParameterizedMessage("Failed to execute bulk
> request {}.", executionId), e);
> listener.afterBulk(executionId, bulkRequest, e);
> } finally {
> if (bulkRequestSetupSuccessful == false) { // if we fail on
> client.bulk() release the semaphore
> toRelease.run();
> }
> }
> }
> {code}
> As the read line i marked above, i think, that's the reason why the retry
> operation thread was block, because the the bulk process thread never release
> the lock on `bulkprocessor`. and, i also trying to figure out why the field
> `concurrentRequests` was set to zero. And i saw the the initialize for
> bulkprocessor in class `ElasticsearchSinkBase`:
> {code:java}
> protected BulkProcessor buildBulkProcessor(BulkProcessor.Listener listener) {
> ...
> BulkProcessor.Builder bulkProcessorBuilder =
> callBridge.createBulkProcessorBuilder(client, listener);
> // This makes flush() blocking
> bulkProcessorBuilder.setConcurrentRequests(0);
>
> ...
> return bulkProcessorBuilder.build();
> }
> {code}
> this field value was set to zero explicitly. So, all things seems to make
> sense, but i still wonder why the retry operation is not in the same thread
> as the bulk process
[jira] [Commented] (FLINK-11046) ElasticSearch6Connector cause thread blocked when index failed with retry
[
https://issues.apache.org/jira/browse/FLINK-11046?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16706611#comment-16706611
]
luoguohao commented on FLINK-11046:
---
sorry for missing that, here are all my settings for the ES sink:
* bulk.flush.max.actions: 1000
* bulk.flush.interval.ms: 10s
* bulk.flush.max.size.mb: 10M
* bulk.flush.backoff.enable: true
* bulk.flush.backoff.retries: 3
* bulk.flush.backoff.type: EXPONENTIAL
* bulk.flush.backoff.delay: 1minute
> ElasticSearch6Connector cause thread blocked when index failed with retry
> -
>
> Key: FLINK-11046
> URL: https://issues.apache.org/jira/browse/FLINK-11046
> Project: Flink
> Issue Type: Bug
> Components: ElasticSearch Connector
>Affects Versions: 1.6.2
>Reporter: luoguohao
>Priority: Major
>
> When i'm using es6 sink to index into es, bulk process with some exception
> catched, and i trying to reindex the document with the call
> `indexer.add(action)` in the `ActionRequestFailureHandler.onFailure()`
> method, but things goes incorrect. The call thread stuck there, and with the
> thread dump, i saw the `bulkprocessor` object was locked by other thread.
> {code:java}
> public interface ActionRequestFailureHandler extends Serializable {
> void onFailure(ActionRequest action, Throwable failure, int restStatusCode,
> RequestIndexer indexer) throws Throwable;
> }
> {code}
> After i read the code implemented in the `indexer.add(action)`, i find that
> `synchronized` is needed on each add operation.
> {code:java}
> private synchronized void internalAdd(DocWriteRequest request, @Nullable
> Object payload) {
> ensureOpen();
> bulkRequest.add(request, payload);
> executeIfNeeded();
> }
> {code}
> And, at i also noticed that `bulkprocessor` object would also locked in the
> bulk process thread.
> the bulk process operation is in the following code:
> {code:java}
> public void execute(BulkRequest bulkRequest, long executionId) {
> Runnable toRelease = () -> {};
> boolean bulkRequestSetupSuccessful = false;
> try {
> listener.beforeBulk(executionId, bulkRequest);
> semaphore.acquire();
> toRelease = semaphore::release;
> CountDownLatch latch = new CountDownLatch(1);
> retry.withBackoff(consumer, bulkRequest, new
> ActionListener() {
> @Override
> public void onResponse(BulkResponse response) {
> try {
> listener.afterBulk(executionId, bulkRequest, response);
> } finally {
> semaphore.release();
> latch.countDown();
> }
> }
> @Override
> public void onFailure(Exception e) {
> try {
> listener.afterBulk(executionId, bulkRequest, e);
> } finally {
> semaphore.release();
> latch.countDown();
> }
> }
> }, Settings.EMPTY);
> bulkRequestSetupSuccessful = true;
>if (concurrentRequests == 0) {
>latch.await();
> }
> } catch (InterruptedException e) {
> Thread.currentThread().interrupt();
> logger.info(() -> new ParameterizedMessage("Bulk request {} has been
> cancelled.", executionId), e);
> listener.afterBulk(executionId, bulkRequest, e);
> } catch (Exception e) {
> logger.warn(() -> new ParameterizedMessage("Failed to execute bulk
> request {}.", executionId), e);
> listener.afterBulk(executionId, bulkRequest, e);
> } finally {
> if (bulkRequestSetupSuccessful == false) { // if we fail on
> client.bulk() release the semaphore
> toRelease.run();
> }
> }
> }
> {code}
> As the read line i marked above, i think, that's the reason why the retry
> operation thread was block, because the the bulk process thread never release
> the lock on `bulkprocessor`. and, i also trying to figure out why the field
> `concurrentRequests` was set to zero. And i saw the the initialize for
> bulkprocessor in class `ElasticsearchSinkBase`:
> {code:java}
> protected BulkProcessor buildBulkProcessor(BulkProcessor.Listener listener) {
> ...
> BulkProcessor.Builder bulkProcessorBuilder =
> callBridge.createBulkProcessorBuilder(client, listener);
> // This makes flush() blocking
> bulkProcessorBuilder.setConcurrentRequests(0);
>
> ...
> return bulkProcessorBuilder.build();
> }
> {code}
> this field value was set to zero explicitly. So, all things seems to make
> sense, but i still wonder why the retry operation is not in the same thread
> as the bulk process execution, after i read the code, `bulkAsync` method
> might be the last puzzle.
> {code:java}
>
[jira] [Commented] (FLINK-11046) ElasticSearch6Connector cause thread blocked when index failed with retry
[
https://issues.apache.org/jira/browse/FLINK-11046?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16706243#comment-16706243
]
miki haiat commented on FLINK-11046:
HI ,
Can you please share the elastic connection properties .
> ElasticSearch6Connector cause thread blocked when index failed with retry
> -
>
> Key: FLINK-11046
> URL: https://issues.apache.org/jira/browse/FLINK-11046
> Project: Flink
> Issue Type: Bug
> Components: ElasticSearch Connector
>Affects Versions: 1.6.2
>Reporter: luoguohao
>Priority: Major
>
> When i'm using es6 sink to index into es, bulk process with some exception
> catched, and i trying to reindex the document with the call
> `indexer.add(action)` in the `ActionRequestFailureHandler.onFailure()`
> method, but things goes incorrect. The call thread stuck there, and with the
> thread dump, i saw the `bulkprocessor` object was locked by other thread.
> {code:java}
> public interface ActionRequestFailureHandler extends Serializable {
> void onFailure(ActionRequest action, Throwable failure, int restStatusCode,
> RequestIndexer indexer) throws Throwable;
> }
> {code}
> After i read the code implemented in the `indexer.add(action)`, i find that
> `synchronized` is needed on each add operation.
> {code:java}
> private synchronized void internalAdd(DocWriteRequest request, @Nullable
> Object payload) {
> ensureOpen();
> bulkRequest.add(request, payload);
> executeIfNeeded();
> }
> {code}
> And, at i also noticed that `bulkprocessor` object would also locked in the
> bulk process thread.
> the bulk process operation is in the following code:
> {code:java}
> public void execute(BulkRequest bulkRequest, long executionId) {
> Runnable toRelease = () -> {};
> boolean bulkRequestSetupSuccessful = false;
> try {
> listener.beforeBulk(executionId, bulkRequest);
> semaphore.acquire();
> toRelease = semaphore::release;
> CountDownLatch latch = new CountDownLatch(1);
> retry.withBackoff(consumer, bulkRequest, new
> ActionListener() {
> @Override
> public void onResponse(BulkResponse response) {
> try {
> listener.afterBulk(executionId, bulkRequest, response);
> } finally {
> semaphore.release();
> latch.countDown();
> }
> }
> @Override
> public void onFailure(Exception e) {
> try {
> listener.afterBulk(executionId, bulkRequest, e);
> } finally {
> semaphore.release();
> latch.countDown();
> }
> }
> }, Settings.EMPTY);
> bulkRequestSetupSuccessful = true;
> if (concurrentRequests == 0) {
> latch.await();
> }
> } catch (InterruptedException e) {
> Thread.currentThread().interrupt();
> logger.info(() -> new ParameterizedMessage("Bulk request {} has been
> cancelled.", executionId), e);
> listener.afterBulk(executionId, bulkRequest, e);
> } catch (Exception e) {
> logger.warn(() -> new ParameterizedMessage("Failed to execute bulk
> request {}.", executionId), e);
> listener.afterBulk(executionId, bulkRequest, e);
> } finally {
> if (bulkRequestSetupSuccessful == false) { // if we fail on
> client.bulk() release the semaphore
> toRelease.run();
> }
> }
> }
> {code}
> As the read line i marked above, i think, that's the reason why the retry
> operation thread was block, because the the bulk process thread never release
> the lock on `bulkprocessor`. and, i also trying to figure out why the field
> `concurrentRequests` was set to zero. And i saw the the initialize for
> bulkprocessor in class `ElasticsearchSinkBase`:
> {code:java}
> protected BulkProcessor buildBulkProcessor(BulkProcessor.Listener listener) {
> ...
> BulkProcessor.Builder bulkProcessorBuilder =
> callBridge.createBulkProcessorBuilder(client, listener);
> // This makes flush() blocking
> bulkProcessorBuilder.setConcurrentRequests(0);
>
> ...
> return bulkProcessorBuilder.build();
> }
> {code}
> this field value was set to zero explicitly. So, all things seems to make
> sense, but i still wonder why the retry operation is not in the same thread
> as the bulk process execution, after i read the code, `bulkAsync` method
> might be the last puzzle.
> {code:java}
> @Override
> public BulkProcessor.Builder createBulkProcessorBuilder(RestHighLevelClient
> client, BulkProcessor.Listener listener) {
> return BulkProcessor.builder(client::bulkAsync, listener);
> }
> {code}
> So, I hope someone can help
