@Override
    public int partition(String topic, Object key, byte[] keyBytes,
            Object value, byte[] valueBytes, Cluster cluster) {
        if (keyBytes == null) {
            return stickyPartitionCache.partition(topic, cluster);
        }
        List<PartitionInfo> partitions = cluster.availablePartitionsForTopic(topic);
        int numPartitions = partitions.size();

        if (numPartitions == 0) {
            partitions = cluster.partitionsForTopic(topic);
            numPartitions = partitions.size();
        }

        // hash the keyBytes to choose a partition
        return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
    }
Compute the partition for the given record.

@param topic      The topic name
@param key        The key to partition on (or null if no key)
@param keyBytes   serialized key to partition on (or null if no key)
@param value      The value to partition on or null
@param valueBytes serialized value to partition on or null
@param cluster    The current cluster metadata
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/kafka/PartitionerSelector.java