Error Handlers

When writing data, the connector batches up documents into a bulk request before sending them to Elasticsearch. In the response, Elasticsearch returns a status for each document sent, which may include rejections or failures. A common error encountered here is a rejection which means that the shard that the document was meant to be written to was too busy to accept the write. Other failures here might include documents that are refused because they do not conform to the current index mapping, or conflict with the current version of the document.

Elasticsearch for Apache Hadoop provides an API to handle document level errors from bulk responses. Error handlers for bulk writes are given:

Always configured as the first error handler for bulk writes.

This handler checks the failures for common HTTP codes that can be retried. These codes are usually indicators that the shard the document would be written to is too busy to accept the write, and the document was rejected to shed load. This handler is always configured to be run first for bulk failures. All handlers that are configured by the user are placed in order after this one.

While the handler’s position in the list of error handlers cannot be modified, its behavior can be modified by adjusting the following configurations:

Default Handler Name: log

When this handler is invoked it logs a message containing the JSON bulk entry that failed, the error message, and any previous handler messages. After logging this message, the handler signals that the error has been acknowledged, thus consuming/ignoring it.

Available configurations for this handler:

Default Handler Name: fail

When this handler is called it rethrows the error given to it and aborts. This handler is always loaded and automatically placed at the end of the list of error handlers.

There are no configurations for this handler.

To configure bulk error handlers, you must specify the handlers in order with the following properties.

For bulk write failures, the HTTP Retry built-in handler is always placed as the first error handler. Additionally, the Abort on Failure built-in handler is always placed as the last error handler to catch any unhandled errors. These two error handlers alone form the default bulk write error handling behavior for elasticsearch-hadoop, which matches the behavior from previous versions.

This behavior is modified by inserting handlers into the chain by using the handlers property. Let’s say that we want to log ALL errors and ignore them.

es.write.rest.error.handlers = log 

With the above configuration, the handler list now looks like the following:

As described above, the built-in log error handler has a required setting: What to use for the logger name. The logger used will respect whatever logging configuration you have in place, and thus needs a name for the logger to use:

es.write.rest.error.handlers = log 
es.write.rest.error.handler.log.logger.name = BulkErrors 

At this point, the Abort on Failure built-in handler is effectively ignored since the Drop and Log built-in handler will always mark an error as consumed. This practice can prove to be hazardous, as potentially important errors may simply be ignored. In many cases, it is preferable for users to write their own error handler to handle expected exceptions.

Let’s say that you are streaming sensitive transaction data to Elasticsearch. In this scenario, your data is carefully versioned and you take advantage of Elasticsearch’s version system to keep from overwriting newer data with older data. Perhaps your data is distributed in a way that allows newer data to sneak in to Elasticsearch before some older bits of data. No worries, the version system will reject the older data and preserve the integrity of the data in Elasticsearch. The problem here is that your streaming job has failed because conflict errors were returned and the connector was unsure if you were expecting that.

Let’s write an error handler for this situation:

package org.myproject.myhandlers;

import org.elasticsearch.hadoop.handler.HandlerResult;
import org.elasticsearch.hadoop.rest.bulk.handler.BulkWriteErrorHandler;
import org.elasticsearch.hadoop.rest.bulk.handler.BulkWriteFailure;
import org.elasticsearch.hadoop.rest.bulk.handler.DelayableErrorCollector;

public class IgnoreConflictsHandler extends BulkWriteErrorHandler { 

    private static final Logger LOGGER = ...; 

    @Override
    public HandlerResult onError(BulkWriteFailure entry, DelayableErrorCollector<byte[]> collector) 
    throws Exception
    {
        if (entry.getResponseCode() == 409) { 
            LOGGER.warn("Encountered conflict response. Ignoring old data.");
            return HandlerResult.HANDLED; 
        }
        return collector.pass("Not a conflict response code."); 
    }
}

Before we can place this handler in the list of bulk write error handlers, we must register the handler class with a name in the settings using es.write.rest.error.handler.[HANDLER-NAME]:

In this case, lets register a handler name for our ignore conflicts handler:

es.write.rest.error.handler.ignoreConflict = org.myproject.myhandlers.IgnoreConflictsHandler

Now that we have a name for the handler, we can use it in the handler list:

es.write.rest.error.handlers = ignoreConflict
es.write.rest.error.handler.ignoreConflict = org.myproject.myhandlers.IgnoreConflictsHandler

Now, your ignore conflict error handler will be invoked whenever a bulk failure occurs, and will instruct the connector that it is ok with ignoring conflict response codes from Elasticsearch.

What if instead of logging data and dropping it, what if you wanted to persist it somewhere for safe keeping? What if we wanted to pass properties into our handlers to parameterize their behavior? Lets create a handler that stores error information in a local file for later analysis.

package org.myproject.myhandlers;

import ...

import org.elasticsearch.hadoop.handler.HandlerResult;
import org.elasticsearch.hadoop.rest.bulk.handler.BulkWriteErrorHandler;
import org.elasticsearch.hadoop.rest.bulk.handler.BulkWriteFailure;
import org.elasticsearch.hadoop.rest.bulk.handler.DelayableErrorCollector;

public class OutputToFileHandler extends BulkWriteErrorHandler { 

    private OutputStream outputStream;   
    private BufferedWriter writer;

    @Override
    public void init(Properties properties) {   
        try {
            outputStream = new FileOutputStream(properties.getProperty("filename"));   
            writer = new BufferedWriter(new OutputStreamWriter(outputStream));
        } catch (FileNotFoundException e) {
            throw new RuntimeException("Could not open file", e);
        }
    }

    @Override
    public HandlerResult onError(BulkWriteFailure entry, DelayableErrorCollector<byte[]> collector)   
    throws Exception
    {
        writer.write("Code: " + entry.getResponseCode());
        writer.newLine();
        writer.write("Error: " + entry.getException().getMessage());
        writer.newLine();
        for (String message : entry.previousHandlerMessages()) {
            writer.write("Previous Handler: " + message);           
            writer.newLine();
        }
        writer.write("Attempts: " + entry.getNumberOfAttempts());
        writer.newLine();
        writer.write("Entry: ");
        writer.newLine();
        IOUtils.copy(entry.getEntryContents(), writer);
        writer.newLine();

        return HandlerResult.HANDLED; 
    }

    @Override
    public void close() {   
        try {
            writer.close();
            outputStream.close();
        } catch (IOException e) {
            throw new RuntimeException("Closing file failed", e);
        }
    }
}

Added to this handler are the init and close methods. The init method is called when the handler is first created at the start of the task and the close method is called when the task concludes. The init method accepts a properties parameter, which contains any handler specific properties set by using es.write.rest.error.handler.[HANDLER-NAME].[PROPERTY-NAME].

In our use case, we will configure the our file logging error handler like so:

es.write.rest.error.handler.writeFile = org.myproject.myhandlers.OutputToFileHandler   
es.write.rest.error.handler.writeFile.filename = /path/to/some/output/file   

Now to bring it all together with the previous example (ignoring conflicts):

es.write.rest.error.handlers = ignoreConflict,writeFile

es.write.rest.error.handler.ignoreConflict = org.myproject.myhandlers.IgnoreConflictsHandler

es.write.rest.error.handler.writeFile = org.myproject.myhandlers.OutputToFileHandler
es.write.rest.error.handler.writeFile.filename = /path/to/some/output/file

You now have a chain of handlers that retries bulk rejections by default (HTTP Retry built-in handler), then ignores any errors that are conflicts (our own ignore conflicts handler), then ignores any other errors by writing them out to a file (our own output to file handler).