Integrating Flink with Elasticsearch: Code Samples and Retry Strategies

1. Introduction to ElasticsearchSink

When processing data with Flink, the final step is to store or export results using a sink. Flink provides several built‑in sink connectors, and this article focuses on the commonly used ElasticsearchSink, covering its basic usage and internal mechanisms.

2. Adding the Maven Dependency

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-elasticsearch2_2.10</artifactId>
  <version>1.3.1</version>
</dependency>

Adjust the version numbers to match the Flink and Elasticsearch versions in use.

3. Basic Implementation Code

DataStream<String> input = ...;

Map<String, String> config = new HashMap<>();
config.put("cluster.name", "my-cluster-name");
// Number of records to batch when writing to ES
config.put("bulk.flush.max.actions", "1");

List<InetSocketAddress> transportAddresses = new ArrayList<>();
transportAddresses.add(new InetSocketAddress(InetAddress.getByName("127.0.0.1"), 9300));
transportAddresses.add(new InetSocketAddress(InetAddress.getByName("10.2.3.1"), 9300));

input.addSink(new ElasticsearchSink<>(config, transportAddresses, new ElasticsearchSinkFunction<String>() {
    public IndexRequest createIndexRequest(String element) {
        Map<String, String> json = new HashMap<>();
        json.put("data", element);
        return Requests.indexRequest()
                .index("my-index")
                .type("my-type")
                .source(json);
    }
    @Override
    public void process(String element, RuntimeContext ctx, RequestIndexer indexer) {
        indexer.add(createIndexRequest(element));
    }
}));

4. Extended Configuration for Retry

When the ES cluster experiences spikes, writes may fail because the default sink lacks a retry mechanism. The following settings enable and tune retries:

// Enable retry mechanism
config.put("bulk.flush.backoff.enable", "true");
// Retry policy type: EXPONENTIAL or CONSTANT
config.put("bulk.flush.backoff.type", "EXPONENTIAL");
// Base delay for exponential backoff (seconds)
config.put("bulk.flush.backoff.delay", "2");
// Number of retry attempts
config.put("bulk.flush.backoff.retries", "3");

Additional options: bulk.flush.max.actions (max records per batch), bulk.flush.max.size.mb (max batch size in MB), bulk.flush.interval.ms (time interval forcing a flush regardless of size).

5. Failure Handler

Write failures caused by a full ES queue or node crashes can be handled by providing an ActionRequestFailureHandler when constructing the sink:

input.addSink(new ElasticsearchSink<>(
    config, transportAddresses,
    new ElasticsearchSinkFunction<String>() {...},
    new ActionRequestFailureHandler() {
        @Override
        void onFailure(ActionRequest action, Throwable failure, int restStatusCode, RequestIndexer indexer) throws Throwable {
            if (ExceptionUtils.containsThrowable(failure, EsRejectedExecutionException.class)) {
                // Re‑queue the failed request for later retry
                indexer.add(action);
            } else if (ExceptionUtils.containsThrowable(failure, ElasticsearchParseException.class)) {
                // Custom handling for parse errors
            } else {
                throw failure;
            }
        }
    }));

If only simple retry is needed, the built‑in RetryRejectedExecutionFailureHandler can be used, which retries on EsRejectedExecutionException .

6. Important Notes

Do not wrap the sink’s process() method in a try‑catch block; the default handler rethrows the exception, making it uncapturable.

To activate the retry mechanism, enable Flink checkpoints with env.enableCheckpoint(). The sink’s retry logic runs during checkpoint flushing, as shown in the snapshotState implementation that checks flushOnCheckpoint and repeatedly calls bulkProcessor.flush() until pending requests are zero.

7. Conclusion

Although ElasticsearchSink implements the CheckpointedFunction interface, it does not use Flink’s state snapshot for recovery; instead, it leverages the checkpoint timing to trigger its own retry loop. This pattern demonstrates how to adapt framework hooks for custom fault‑tolerance logic.