private int partition(ProducerRecord<K, V> record, byte[] serializedKey, byte[] serializedValue, Cluster cluster) {
        if (record.partition() != null)
            return record.partition();

        if (partitioner != null) {
            int customPartition = partitioner.partition(
                record.topic(), record.key(), serializedKey, record.value(), serializedValue, cluster);
            if (customPartition < 0) {
                throw new IllegalArgumentException(String.format(
                    "The partitioner generated an invalid partition number: %d. Partition number should always be non-negative.", customPartition));
            }
            return customPartition;
        }

        if (serializedKey != null && !partitionerIgnoreKeys) {
            // hash the keyBytes to choose a partition
            return BuiltInPartitioner.partitionForKey(serializedKey, cluster.partitionsForTopic(record.topic()).size());
        } else {
            return RecordMetadata.UNKNOWN_PARTITION;
        }
    }
computes partition for given record.
if the record has partition returns the value otherwise
if custom partitioner is specified, call it to compute partition
otherwise try to calculate partition based on key.
If there is no key or key should be ignored return
RecordMetadata.UNKNOWN_PARTITION to indicate any partition
can be used (the partition is then calculated by built-in
partitioning logic).
clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java