Shengnan YU created FLINK-14938:
-----------------------------------
Summary: Flink elasticsearch failure handler re-add indexrequest causes ConcurrentModificationException
Key: FLINK-14938
URL:
https://issues.apache.org/jira/browse/FLINK-14938 Project: Flink
Issue Type: Bug
Components: Connectors / ElasticSearch
Affects Versions: 1.8.1
Reporter: Shengnan YU
When use Elasticsearch connector failure handler (from official example) to re-add documents, Flink encountered ConcurrentModificationException.
{code:java}
input.addSink(new ElasticsearchSink<>(
config, transportAddresses,
new ElasticsearchSinkFunction<String>() {...},
new ActionRequestFailureHandler() {
@Override
void onFailure(ActionRequest action,
Throwable failure,
int restStatusCode,
RequestIndexer indexer) throw Throwable {
if (ExceptionUtils.findThrowable(failure, EsRejectedExecutionException.class).isPresent()) {
// full queue; re-add document for indexing
indexer.add(action);
}
}
}));
{code}
I found that in method BufferingNoOpRequestIndexer$processBufferedRequests, it will iterator a list of ActionRequest. However the failure handler will keep re-adding request to that list after bulk, which causes ConcurrentModificationException.
I think it should be a multi-thread bug and need to find a thread-safe List to maintain the failure request?
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
Locked
