apache / apache/beam

[Bug]: Java ElasticsearchIO fails to handle retry on HTTP 429

Open
#22,160 5 comments 0 reactions 1 assignee Claimed by @egalpin View on GitHub
bug elasticsearch io java P2
Dominant language
Java
Stars
8.7k
Forks
4.7k
Avg merge
1d 20h
Merged PRs (30d)
196

Description

### What happened?

It seems to me that the retry functionality in ElasticsearchIO does not work in my batch pipeline running on Dataflow. Here is an extract from my logs:
```
org.apache.beam.sdk.util.UserCodeException: org.elasticsearch.client.ResponseException: method [POST], host [http://10.x.y.z:9200], URI [/my_index/_bulk], status line [HTTP/1.1 429 Too Many Requests]
{"error":{"root_cause":[{"type":"es_rejected_execution_exception","reason":"rejected execution of coordinating operation [coordinating_and_primary_bytes=663765981, replica_bytes=0, all_bytes=663765981, coordinating_operation_bytes=47402063, max_coordinating_and_primary_bytes=675282944]"}],"type":"es_rejected_execution_exception","reason":"rejected execution of coordinating operation [coordinating_and_primary_bytes=663765981, replica_bytes=0, all_bytes=663765981, coordinating_operation_bytes=47402063, max_coordinating_and_primary_bytes=675282944]"},"status":429}
at org.apache.beam.sdk.util.UserCodeException.wrap(UserCodeException.java:39)
at org.apache.beam.sdk.io.elasticsearch.ElasticsearchIO$BulkIO$BulkIOBundleFn$DoFnInvoker.invokeProcessElement(Unknown Source)
at org.apache.beam.fn.harness.FnApiDoFnRunner.processElementForParDo(FnApiDoFnRunner.java:779)
at org.apache.beam.fn.harness.data.PCollectionConsumerRegistry$MetricTrackingFnDataReceiver.accept(PCollectionConsumerRegistry.java:255)
at org.apache.beam.fn.harness.data.PCollectionConsumerRegistry$MetricTrackingFnDataReceiver.accept(PCollectionConsumerRegistry.java:209)
at org.apache.beam.fn.harness.MapFnRunners$ExplodedWindowedValueMapperFactory.lambda$create$0(MapFnRunners.java:164)
at org.apache.beam.fn.harness.data.PCollectionConsumerRegistry$MetricTrackingFnDataReceiver.accept(PCollectionConsumerRegistry.java:255)
at org.apache.beam.fn.harness.data.PCollectionConsumerRegistry$MetricTrackingFnDataReceiver.accept(PCollectionConsumerRegistry.java:209)
at org.apache.beam.fn.harness.FnApiDoFnRunner.outputTo(FnApiDoFnRunner.java:1769)
at org.apache.beam.fn.harness.FnApiDoFnRunner.access$2800(FnApiDoFnRunner.java:142)
at org.apache.beam.fn.harness.FnApiDoFnRunner$NonWindowObservingProcessBundleContext.outputWithTimestamp(FnApiDoFnRunner.java:2338)
at org.apache.beam.fn.harness.FnApiDoFnRunner$ProcessBundleContextBase.output(FnApiDoFnRunner.java:2508)
at org.apache.beam.sdk.io.elasticsearch.ElasticsearchIO$DocToBulk$DocToBulkFn.processElement(ElasticsearchIO.java:1638)
at org.apache.beam.sdk.io.elasticsearch.ElasticsearchIO$DocToBulk$DocToBulkFn$DoFnInvoker.invokeProcessElement(Unknown Source)
<>
Caused by: org.elasticsearch.client.ResponseException: method [POST], host [http://10.x.y.z:9200], URI [/my_index/_bulk], status line [HTTP/1.1 429 Too Many Requests]
{"error":{"root_cause":[{"type":"es_rejected_execution_exception","reason":"rejected execution of coordinating operation [coordinating_and_primary_bytes=663765981, replica_bytes=0, all_bytes=663765981, coordinating_operation_bytes=47402063, max_coordinating_and_primary_bytes=675282944]"}],"type":"es_rejected_execution_exception","reason":"rejected execution of coordinating operation [coordinating_and_primary_bytes=663765981, replica_bytes=0, all_bytes=663765981, coordinating_operation_bytes=47402063, max_coordinating_and_primary_bytes=675282944]"},"status":429}
at org.elasticsearch.client.RestClient.convertResponse(RestClient.java:346)
at org.elasticsearch.client.RestClient.performRequest(RestClient.java:312)
at org.elasticsearch.client.RestClient.performRequest(RestClient.java:287)
at org.apache.beam.sdk.io.elasticsearch.ElasticsearchIO$BulkIO$BulkIOBaseFn.flushBatch(ElasticsearchIO.java:2502)
at org.apache.beam.sdk.io.elasticsearch.ElasticsearchIO$BulkIO$BulkIOBaseFn.flushAndOutputResults(ElasticsearchIO.java:2435)
at org.apache.beam.sdk.io.elasticsearch.ElasticsearchIO$BulkIO$BulkIOBaseFn.addAndMaybeFlush(ElasticsearchIO.java:2448)
at org.apache.beam.sdk.io.elasticsearch.ElasticsearchIO$BulkIO$BulkIOBundleFn.processElement(ElasticsearchIO.java:2333)
```

The Elasticsearch `RestClient` throws an exception when the response status is an error, unless a special HTTP request parameter called "ignore" is defined and includes the status code in question.

I've created this reproducer:
* Code: https://gitlab.com/joakim.edenholm/apache-beam-repro/-/tree/elastic-throttling
* CI logs: https://gitlab.com/joakim.edenholm/apache-beam-repro/-/jobs/2679857923

### Issue Priority

Priority: 2

### Issue Component

Component: io-java-elasticsearch

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.