@Override
        public int compare(Tuple2<String, Long> k1, Tuple2<String, Long> k2) {
            if (!Objects.equals(k1.f0, k2.f0)) {
                return Comparator.nullsLast(Comparator.comparing((Tuple2<String, Long> k) -> k.f0)).compare(k1, k2);
            }
            return Comparator.comparing((Tuple2<String, Long> k) -> k.f1).compare(k1, k2);
        }
A {@link Convertor} that key represent <partition, serial number>
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/collections/PartitionGroupBuffer.java