You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Robert Metzger (Jira)" <ji...@apache.org> on 2019/12/12 13:26:00 UTC
[jira] [Commented] (FLINK-14938) Flink elasticsearch failure
handler re-add indexrequest causes ConcurrentModificationException
[ https://issues.apache.org/jira/browse/FLINK-14938?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16994661#comment-16994661 ]
Robert Metzger commented on FLINK-14938:
----------------------------------------
Thanks for looking into this. I've assigned you!
> 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
> Assignee: Shengnan YU
> Priority: Major
>
>
> 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.
> {code:java}
> void processBufferedRequests(RequestIndexer actualIndexer) {
> for (ActionRequest request : bufferedRequests) {
> if (request instanceof IndexRequest) {
> actualIndexer.add((IndexRequest) request);
> } else if (request instanceof DeleteRequest) {
> actualIndexer.add((DeleteRequest) request);
> } else if (request instanceof UpdateRequest) {
> actualIndexer.add((UpdateRequest) request);
> }
> }
> bufferedRequests.clear();
> }{code}
> I think it should be a multi-thread bug and is it ok to use concurrent queue to maintain the failure request?
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)