@Override
        public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
            try {
                for (DocWriteRequest<?> writeRequest : request.requests()) {
                    getFailureHandler().onFailure(writeRequest, failure, -1, getFailureRequestIndexer());
                }
            } catch (Throwable t) {
                // fail the sink and skip the rest of the items
                // if the failure handler decides to throw an exception
                getFailureThrowable().compareAndSet(null, t);
            }
            if (flushOnCheckpoint()) {
                getPendingRequests().getAndAdd(-request.numberOfActions());
            }
        }
Creates the Elasticsearch sink.
Use {@link DirtySinkFailureHandler} when need sink dirty data

@return the created Elasticsearch sink.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/elasticsearch-7/src/main/java/org/apache/inlong/sort/elasticsearch7/ElasticsearchSink.java