@Override
        public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
            try {
                for (DocWriteRequest writeRequest : request.requests()) {
                    if (writeRequest instanceof ActionRequest) {
                        failureHandler.onFailure(
                                (ActionRequest) writeRequest, failure, -1, failureRequestIndexer);
                    } else {
                        throw new UnsupportedOperationException(
                                "The sink currently only supports ActionRequests");
                    }
                }
            } catch (Throwable t) {
                // fail the sink and skip the rest of the items
                // if the failure handler decides to throw an exception
                failureThrowable.compareAndSet(null, t);
            }

            if (flushOnCheckpoint) {
                numPendingRequests.getAndAdd(-request.numberOfActions());
            }
        }
Build the {@link BulkProcessor}.

<p>Note: this is exposed for testing purposes.
inlong-sort/sort-flink/sort-flink-v1.18/sort-connectors/elasticsearch-base/src/main/java/org/apache/inlong/sort/elasticsearch/ElasticsearchSinkBase.java