public static void checkSchema(
            String[] fieldNames,
            DataType[] dataTypes,
            Schema kuduTableSchema) {
        List<ColumnSchema> kuduColumns = kuduTableSchema.getColumns();
        Map<String, ColumnSchema> columnMap = kuduColumns
                .stream()
                .map(col -> Tuple2.of(col.getName(), col))
                .collect(Collectors.toMap(t2 -> t2.f0, t2 -> t2.f1));
        String missFields = Arrays
                .stream(fieldNames)
                .filter(s -> !columnMap.containsKey(s))
                .collect(Collectors.joining(","));
        checkState(StringUtils.isNullOrWhitespaceOnly(missFields), "Can not find fields {} in kudu table!", missFields);
    }
Check schema according to field name
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/kudu/src/main/java/org/apache/inlong/sort/kudu/common/KuduUtils.java