@SuppressWarnings("unchecked")
    private void addDocument(RowData element, String key, String index, byte[] document,
            RequestIndexer<Request> indexer) {
        Request request;
        switch (element.getRowKind()) {
            case INSERT:
            case UPDATE_AFTER:
                if (key != null) {
                    request = requestFactory.createUpdateRequest(index, docType, key, contentType, document);
                    if (addRouting(request, element, document)) {
                        indexer.add(request);
                        sendMetrics(document, index);
                    }
                } else {
                    request = requestFactory.createIndexRequest(index, docType, key, contentType, document);
                    if (addRouting(request, element, document)) {
                        indexer.add(request);
                        sendMetrics(document, index);
                    }
                }
                break;
            case DELETE:
                request = requestFactory.createDeleteRequest(index, docType, key);
                if (addRouting(request, element, document)) {
                    indexer.add(request);
                    sendMetrics(document, index);
                }
                break;
            case UPDATE_BEFORE:
                sendMetrics(document, index);
                break;
            default:
                LOGGER.error(String.format("The type of element should be 'RowData' only, raw data: %s", element));
                handleDirty(element, DirtyType.UNSUPPORTED_DATA_TYPE,
                        new RuntimeException("The type of element should be 'RowData' only."), index);
        }
    }
Sink function for converting upserts into Elasticsearch ActionRequests.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/elasticsearch-base/src/main/java/org/apache/inlong/sort/elasticsearch/table/MultipleElasticsearchSinkFunctionBase.java