Elastic search sink error handling

[Nicholas Walton]

HI,

I need help with handling errors with the elasticsearch sink as below

2019-11-19 08:09:09,043 ERROR org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkBase  - Failed Elasticsearch item request: [flink-index-deduplicated/nHWQM0XMSTatRri7zw_s_Q][[flink-index-deduplicated][13]] ElasticsearchException[Elasticsearch exception [type=version_conflict_engine_exception, reason=[75:108]: version conflict, document already exists (current version [1])]]

[flink-index-deduplicated/nHWQM0XMSTatRri7zw_s_Q][[flink-index-deduplicated][13]] ElasticsearchException[Elasticsearch exception [type=version_conflict_engine_exception, reason=[75:108]: version conflict, document already exists (current version [1])]]

at org.elasticsearch.ElasticsearchException.innerFromXContent(ElasticsearchException.java:510)

at org.elasticsearch.ElasticsearchException.fromXContent(ElasticsearchException.java:421)

at org.elasticsearch.action.bulk.BulkItemResponse.fromXContent(BulkItemResponse.java:135)

at org.elasticsearch.action.bulk.BulkResponse.fromXContent(BulkResponse.java:198)

at org.elasticsearch.client.RestHighLevelClient.parseEntity(RestHighLevelClient.java:653)

at org.elasticsearch.client.RestHighLevelClient.lambda$performRequestAsyncAndParseEntity$3(RestHighLevelClient.java:549)

at org.elasticsearch.client.RestHighLevelClient$1.onSuccess(RestHighLevelClient.java:580)

at org.elasticsearch.client.RestClient$FailureTrackingResponseListener.onSuccess(RestClient.java:621)

at org.elasticsearch.client.RestClient$1.completed(RestClient.java:375)

at org.elasticsearch.client.RestClient$1.completed(RestClient.java:366)

at org.apache.http.concurrent.BasicFuture.completed(BasicFuture.java:122)

at org.apache.http.impl.nio.client.DefaultClientExchangeHandlerImpl.responseCompleted(DefaultClientExchangeHandlerImpl.java:177)

at org.apache.http.nio.protocol.HttpAsyncRequestExecutor.processResponse(HttpAsyncRequestExecutor.java:436)

at org.apache.http.nio.protocol.HttpAsyncRequestExecutor.inputReady(HttpAsyncRequestExecutor.java:326)

at org.apache.http.impl.nio.DefaultNHttpClientConnection.consumeInput(DefaultNHttpClientConnection.java:265)

at org.apache.http.impl.nio.client.InternalIODispatch.onInputReady(InternalIODispatch.java:81)

at org.apache.http.impl.nio.client.InternalIODispatch.onInputReady(InternalIODispatch.java:39)

at org.apache.http.impl.nio.reactor.AbstractIODispatch.inputReady(AbstractIODispatch.java:114)

at org.apache.http.impl.nio.reactor.BaseIOReactor.readable(BaseIOReactor.java:162)

at org.apache.http.impl.nio.reactor.AbstractIOReactor.processEvent(AbstractIOReactor.java:337)

at org.apache.http.impl.nio.reactor.AbstractIOReactor.processEvents(AbstractIOReactor.java:315)

at org.apache.http.impl.nio.reactor.AbstractIOReactor.execute(AbstractIOReactor.java:276)

at org.apache.http.impl.nio.reactor.BaseIOReactor.execute(BaseIOReactor.java:104)

at org.apache.http.impl.nio.reactor.AbstractMultiworkerIOReactor$Worker.run(AbstractMultiworkerIOReactor.java:588)

at java.lang.Thread.run(Thread.java:748)

The error is expected since I  am creating documents with duplicate ids, so I can only load new data from a previous batch that was only partially loaded or due to a timeout I’ve uploaded a document twice to ensure the document is definitely loaded and not lost in the timeout. 

The document is created as

          val json = new util.HashMap[String, Any]

          json.put("arrayinstance", esid)

          json.put("bearing", element._1)

          json.put("sampleindex", element._2)

          json.put("sample", element._3)

          json.put("hashstring", element._4)

          json.put("priorrepeats", element._5)

          return Requests.indexRequest()

            .index("flink-index-deduplicated")

            .`type`("_doc")

            .id(element._1+":"+element._2)

            .create(true)

            .source(json)

        }

My problem is how can I catch the failure, recover and carryon? I have set a failure handler as below which will need extending to handle the failure above

    esSinkBuilder.setFailureHandler(

      new ActionRequestFailureHandler() {

        @throws(classOf[Throwable])

        @Override

        override def onFailure(action: ActionRequest, failure: Throwable, restStatusCode: Int, indexer: RequestIndexer) {

          if (ExceptionUtils.findThrowable(failure, classOf[EsRejectedExecutionException]).isPresent) {

            Job.LOG.info("ElasticSearch full queue; re-added document for indexing")

            indexer.add(action)

          } else if (ExceptionUtils.findThrowable(failure, classOf[ElasticsearchParseException]).isPresent) {

            LOG.info("Malformed ElasticSearch document. Document dropped")

          } else if (ExceptionUtils.findThrowable(failure, classOf[java.net.SocketTimeoutException]).isPresent) {

            LOG.info("ElasticSearch document timeout; re-added document for indexing")

            indexer.add(action)

          }/* else if (ExceptionUtils.findThrowable(failure, classOf[]).isPresent) {

            LOG.info("ElasticSearch document duplicate; ignored document")

          } */else {

            // for all other failures, fail the sink

            // here the failure is simply rethrown, but users can also choose to throw custom exceptions

            Job.LOG.info(failure.getMessage)

            throw failure

          }

        }

      }

    )

 

I have tried just ignoring the failure by removing the "throw failure” but to no avail