in flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/RowElasticsearchSinkFunction.java [78:91]
public void process(RowData element, RuntimeContext ctx, RequestIndexer indexer) {
switch (element.getRowKind()) {
case INSERT:
case UPDATE_AFTER:
processUpsert(element, indexer);
break;
case UPDATE_BEFORE:
case DELETE:
processDelete(element, indexer);
break;
default:
throw new TableException("Unsupported message kind: " + element.getRowKind());
}
}