Martijn Visser created FLINK-30526: -------------------------------------- Summary: Handle failures in OpenSearch with ActionRequestFailureHandler being deprecated Key: FLINK-30526 URL: https://issues.apache.org/jira/browse/FLINK-30526 Project: Flink Issue Type: Bug Components: Connectors / Opensearch Reporter: Martijn Visser
{quote} Hi everyone, I have a streaming application that has Elasticsearch sink. I Upgraded flink version from 1.11 to 1.16 and also moved from ES 7 to OpenSearch 2.0, and now I'm facing some deprected issues, hope you can help me. In the previous version I created ElasticsearchSink and added a failure handler, which protected the sink to not fail on some exceptions. final ActionRequestFailureHandler failureHandler = (action, failure, restStatusCode, indexer) -> { if (ExceptionUtils.findThrowable(failure, EsRejectedExecutionException.class).isPresent()) { indexer.add(action); } else if (ExceptionUtils.findThrowable(failure, ElasticsearchParseException.class).isPresent()) { log.warn("Got malformed document , action {}", action); // malformed document; simply drop elasticsearchSinkFunction without failing sink } else if (failure instanceof IOException && failure.getCause() instanceof NullPointerException && failure.getMessage().contains("Unable to parse response body")) { //issue with ES 7 and opensearch - that does not send type - while response is waiting for it //at org.elasticsearch.action.DocWriteResponse.<init>(DocWriteResponse.java:127) -- this.type = Objects.requireNonNull(type); log.debug("known issue format the response for ES 7.5.1 and DB OS (opensearch) :{}", failure.getMessage()); } else { // for all other failures, log and don't fail the sink log.error("Got error while trying to perform ES action {}", action, failure); } }; final ElasticsearchSink.Builder<T> builder = new ElasticsearchSink.Builder<>(transportNodes, elasticsearchSinkFunction); In the new version the class ActionRequestFailureHandler is deprecated and after investigation I can't find any way to handle failures. For all failures the sink fails. Is there anything I didn't see? Thanks is advance! {quote} >From the Apache Flink Slack channel >https://apache-flink.slack.com/archives/C03G7LJTS2G/p1672122873318899 -- This message was sent by Atlassian Jira (v8.20.10#820010)