@Override
    public void onFailure(
            ActionRequest action, Throwable failure, int restStatusCode, RequestIndexer indexer)
            throws Throwable {
        LOG.error("Failed Elasticsearch item request: {}", failure.getMessage(), failure);
        if (ExceptionUtils.findThrowable(failure, EsRejectedExecutionException.class).isPresent()) {
            indexer.add(action);
        } else {
            // rethrow all other failures
            throw failure;
        }
    }
An {@link ActionRequestFailureHandler} that re-adds requests that failed due to temporary {@link
EsRejectedExecutionException}s (which means that Elasticsearch node queues are currently full),
and fails for all other failures.
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/elasticsearch-base/src/main/java/org/apache/inlong/sort/elasticsearch/util/RetryRejectedExecutionFailureHandler.java