Currently, the Elasticsearch Sink actually doesn't offer any guarantees for message delivery.
For proper support of at-least-once, the sink will need to participate in Flink's checkpointing: when snapshotting is triggered at the ElasticsearchSink, we need to synchronize on the pending ES requests by flushing the internal bulk processor. For temporary ES failures (see
FLINK-5122) that may happen on the flush, we should retry them before returning from snapshotting and acking the checkpoint. If there are non-temporary ES failures on the flush, the current snapshot should fail.