@Override
    public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
        LOG.debug("afterBulk,executionId,executionId:{},request:{},response:{}", executionId, request, response);
        BulkItemResponse[] itemResponses = response.getItems();
        List<DocWriteRequest<?>> requests = request.requests();
        if (itemResponses.length != requests.size()) {
            LOG.error("BulkItemResponse size is not equal to IndexRequest size:requestSize:{},responseSize:{}",
                    requests.size(), itemResponses.length);
        }
        for (int i = 0; i < itemResponses.length; i++) {
            // parameter
            EsIndexRequest requestItem = (EsIndexRequest) requests.get(i);
            BulkItemResponse responseItem = itemResponses[i];
            ProfileEvent event = requestItem.getEvent();
            long sendTime = requestItem.getSendTime();

            // is fail
            if (responseItem.isFailed()) {
                context.addSendResultMetric(event, context.getTaskName(), false, sendTime);
                context.backDispatchQueue(requestItem);
            } else {
                context.addSendResultMetric(event, context.getTaskName(), true, sendTime);
                context.releaseDispatchQueue(requestItem);
                event.ack();
            }
        }
    }
afterBulk

@param executionId
@param request
@param response
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/elasticsearch/EsCallbackListener.java