public static Function<RowData, String> createRoutingExtractor(
            TableSchema schema,
            @Nullable String filedName) {
        if (filedName == null) {
            return null;
        }
        List<TableColumn> tableColumns = schema.getTableColumns();
        for (int i = 0; i < schema.getFieldCount(); i++) {
            TableColumn column = tableColumns.get(i);
            if (column.getName().equals(filedName)) {
                RowData.FieldGetter fieldGetter = RowData.createFieldGetter(
                        column.getType().getLogicalType(),
                        i);
                return (Function<RowData, String> & Serializable) (row) -> {
                    Object fieldOrNull = fieldGetter.getFieldOrNull(row);
                    if (fieldOrNull != null) {
                        return fieldOrNull.toString();
                    } else {
                        return null;
                    }
                };
            }
        }
        throw new IllegalArgumentException("Filed " + filedName + " not exist in table schema.");
    }
An extractor for a Elasticsearch routing from a {@link RowData}.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/elasticsearch-base/src/main/java/org/apache/inlong/sort/elasticsearch/table/RoutingExtractor.java