@Override
        public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
            if (response.hasFailures()) {
                BulkItemResponse itemResponse;
                Throwable failure;
                int restStatus;
                try {
                    for (int i = 0; i < response.getItems().length; i++) {
                        itemResponse = response.getItems()[i];
                        failure = getCallBridge().extractFailureCauseFromBulkItemResponse(itemResponse);
                        if (failure != null) {
                            restStatus = itemResponse.getFailure().getStatus() != null ? itemResponse.getFailure()
                                    .getStatus().getStatus() : -1;
                            getFailureHandler().onFailure(request.requests().get(i),
                                    failure, restStatus, 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