private void compareKeyFields(String[] plannerKeyFields, String[] userKeyFields) {
        if (plannerKeyFields == null || plannerKeyFields.length == 0) {
            return;
        }
        if (userKeyFields == null || userKeyFields.length == 0) {
            return;
        }

        Set<String> assumedSet = new HashSet<>(Arrays.asList(plannerKeyFields));
        Set<String> userSet = new HashSet<>(Arrays.asList(userKeyFields));

        if (!assumedSet.equals(userSet)) {
            String errorMsg = String.format(
                    "Key fields provided by flink [%s] are not the same as key fields " +
                            "provided by user [%s]. Please adjust your key fields settings, or " +
                            "set %s to false.",
                    assumedSet, userSet, KuduOptions.ENABLE_KEY_FIELD_CHECK.key());
            throw new ValidationException(errorMsg);
        }
    }
Compare key fields given by flink planner and key fields specified by user.

@param plannerKeyFields Key fields given by flink planner.
@param userKeyFields Key fields specified by user via {@link KuduOptions#SINK_KEY_FIELD_NAMES}.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/kudu/src/main/java/org/apache/inlong/sort/kudu/table/KuduDynamicTableSink.java