@Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        int nextValue = nextValue(topic);
        List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);
        if (!availablePartitions.isEmpty()) {
            int part = Utils.toPositive(nextValue) % availablePartitions.size();
            return availablePartitions.get(part).partition();
        } else {
            // no partitions are available, give a non-available partition
            int numPartitions = cluster.partitionsForTopic(topic).size();
            return Utils.toPositive(nextValue) % 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
clients/src/main/java/org/apache/kafka/clients/producer/RoundRobinPartitioner.java