Hi all, I have a pipeline that is currently reading from Kafka and writing to Elasticsearch. I recently was doing some testing for how it handles failures and was wondering if there’s a best practice or recommendation for doing so. I know that previously the now deprecated

ElasticsearchSink.Builder

class exposed an

onFailure

handler to capture these types of situations:

builder.onFailureHander { request, throwable, statusCode, indexer -> 
    // Handle failures here
}

However, after the sink was recently updated to use the

Elasticsearch7SinkBuilder

class, it doesn’t appear to expose the same level of flexibility for capturing failures within the sink outside of adjusting the backoff strategy. Ideally, I’d need the flexibility to process records in batches (via the internal

BulkProcessor

) with the ability to discern between the types of failures that might occur similar to the following example:

builder.onFailureHandler { request, throwable, statusCode, indexer -> 
    if (ExceptionUtils.findThrowable(throwable, EsRejectedExecutionException.class).isPresent()) {
        // Potentially transient/ES-related error, retry indexing
        indexer.add(action);
    } else if (ExceptionUtils.findThrowable(throwable, ElasticsearchParseException.class).isPresent()) {
        // Bad/invalid document, need to log and potentially send to a DLQ for further investigation
    } else {
        // Anything else, crash the job via an exception
   } 
}

Additionally, when processing in batches, how are single failures within a

BulkProcessor

call typically handled? Is the entire batch retried, or does the processor only reattempt the requests that initially failed. For example, if I had a single request that attempted to index 100 documents, but one of them was a duplicate or could not be inserted into the index, how would that be handled? Ideally, I’d like to capture the document(s) that couldn’t’ be written (i.e. throw them to a DLQ or something) or have some way to isolate what failed in the batch.

Hi Rion, I am looking into exactly the same thing, how to handle failures and gracefully handle exceptions with this sink, but reached the same conclusions as you. By any chance, did you found a decent solution in the meantime? Thanks.

Hi @Tudor Plugaru , Not necessarily an elegant one. I eventually continued to use the older version of the sink that exposed the addition

onFailure

handler and added several conditional statements to capture the types of exceptions that were occurring and what I wanted to do with each (i.e. explicitly retrying transient issues. Something like this:

builder.setFailureHandler { actionRequest, throwable, _, indexer ->
            val request = actionRequest as IndexRequest
            // All of these errors are only thrown after the default number of retries
            when (val cause = getCauseFromThrowable(throwable)) {
                ElasticsearchExceptionTypes.SOCKET_TIMEOUT_EXCEPTION,
                ElasticsearchExceptionTypes.ES_REJECTED_EXECUTION_EXCEPTION,
                ElasticsearchExceptionTypes.CONNECTION_EXCEPTION -> {
                    // Transient exceptions, retry
                    indexer.add(actionRequest)
                }
                
                ElasticsearchExceptionTypes.VERSION_CONFLICT_ENGINE_EXCEPTION -> {
                    // Duplicate, log and don't retry
                }
                ElasticsearchExceptionTypes.PARSE_EXCEPTION -> {
                    // Bad message, will never succeed, log, don't retry, and consider DLQ
                }
                else -> {
                    // Something else bad happened, throw the exception/kill the job
                    throw throwable
                }
            }
 }

Thanks @Rion Williams for that. Good to have the confirmation that there is no way to do it with the new sink.