private List<Integer> getFieldPos() {
        // the positions of the partition keys.
        List<Integer> positions = new ArrayList<>();
        if (partitionKey == null) {
            LOG.error("primaryKeyPartioner:failed to fetch partitionKey");
            return positions;
        }
        // a map storing field names and their position in the schema
        HashMap<String, Integer> map = new HashMap<>();
        String[] keys = partitionKey.split(",");
        for (int i = 0; i < fieldNames.length; i++) {
            map.put(fieldNames[i], i);
        }
        for (String key : keys) {
            positions.add(map.get(key));
        }
        return positions;
    }
output: integer list of partitionKeys' corresponding positions within schema
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/kafka/src/main/java/org/apache/inlong/sort/kafka/partitioner/SingleTableCustomFieldsPartitioner.java