private void handleDirty(RowData rowData, DirtyType dirtyType, Exception e, String index) {
        // skip the index in which the error has occurred
        if (SchemaUpdateExceptionPolicy.STOP_PARTIAL == schemaUpdateExceptionPolicy) {
            if (index != null) {
                errorSet.add(index);
            } else {
                return;
            }
        }

        // keep retry the entire task until it succeeds
        if (SchemaUpdateExceptionPolicy.THROW_WITH_STOP == schemaUpdateExceptionPolicy) {
            throw new RuntimeException(String.format("Writing records %s failed, restarting task",
                    rowData), e);
        }

        // dirty data & archive
        if (SchemaUpdateExceptionPolicy.LOG_WITH_IGNORE == schemaUpdateExceptionPolicy) {
            dirtySinkHelper.invoke(rowData, dirtyType, e);
            if (sinkMetricData == null) {
                return;
            }

            if (index != null) {
                sinkMetricData.outputDirtyMetrics(index, 1,
                        rowData.toString().getBytes(StandardCharsets.UTF_8).length);
            } else {
                sinkMetricData.invokeDirty(1, rowData.toString().getBytes(StandardCharsets.UTF_8).length);
            }
        }
    }
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