[Bug]: Java ElasticsearchIO fails to handle retry on HTTP 429
- 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
Assessment
This issue has not been assessed yet.