@Override
    public void onFailure(DocWriteRequest<?> request, Throwable failure, int restStatusCode,
            RequestIndexer<DocWriteRequest<?>> indexer) throws Throwable {
        // We think that only when ElasticsearchException occurs, it is considered dirty data
        if (ExceptionUtils.findThrowable(failure, ElasticsearchException.class).isPresent()) {
            String dirtyData;
            if (request instanceof UpdateRequest) {
                dirtyData = ((UpdateRequest) request).doc().source().utf8ToString();
            } else if (request instanceof IndexRequest) {
                dirtyData = ((IndexRequest) request).source().utf8ToString();
            } else {
                dirtyData = request.id();
            }
            LOGGER.error("Action request handle error", failure);
            LOGGER.debug("The dirty data: {}", dirtyData);
            dirtySinkHelper.invoke(dirtyData, DirtyType.DOCUMENT_PARSE_ERROR, failure);
            if (sinkMetricData != null) {
                sinkMetricData.invokeDirty(1, dirtyData.getBytes(StandardCharsets.UTF_8).length);
            }
        } else {
            throw failure;
        }
    }
Dirty data sink when failure happens
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/elasticsearch-7/src/main/java/org/apache/inlong/sort/elasticsearch7/utils/DirtySinkFailureHandler.java