public void applyRow(RowData row) throws KuduException {

        Operation operation;
        if (row.getRowKind() == RowKind.UPDATE_AFTER) {
            operation = forceInUpsertMode ? kuduTable.newUpsert() : kuduTable.newInsert();
        } else {
            operation = kuduTable.newDelete();
        }
        final PartialRow kuduRow = operation.getRow();

        for (int j = 0; j < fieldNames.length; j++) {
            String columnName = fieldNames[j];
            DataType columnType = dataTypes[j];

            Object column = RowData.createFieldGetter(columnType.getLogicalType(), j).getFieldOrNull(row);
            if (column == null) {
                continue;
            }
            kuduRow.addObject(columnName, column);
        }
        session.apply(operation);
        applyNum++;
    }
The exception caught in asynchronous tasks.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/kudu/src/main/java/org/apache/inlong/sort/kudu/sink/KuduWriter.java