@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;
        }
    }
@Test
    public void testRoundRobinWithUnavailablePartitions() {
        // Intentionally make the partition list not in partition order to test the edge
        // cases.
        List<PartitionInfo> partitions = asList(
                new PartitionInfo("test", 1, null, NODES, NODES),
                new PartitionInfo("test", 2, NODES[1], NODES, NODES),
                new PartitionInfo("test", 0, NODES[0], NODES, NODES));
        // When there are some unavailable partitions, we want to make sure that (1) we
        // always pick an available partition,
        // and (2) the available partitions are selected in a round robin way.
        int countForPart0 = 0;
        int countForPart2 = 0;
        Partitioner partitioner = new RoundRobinPartitioner();
        Cluster cluster = new Cluster("clusterId", asList(NODES[0], NODES[1], NODES[2]), partitions,
            Collections.emptySet(), Collections.emptySet());
        for (int i = 1; i <= 100; i++) {
            int part = partitioner.partition("test", null, null, null, null, cluster);
            assertTrue(part == 0 || part == 2, "We should never choose a leader-less node in round robin");
            if (part == 0)
                countForPart0++;
            else
                countForPart2++;
        }
        assertEquals(countForPart0, countForPart2, "The distribution between two available partitions should be even");
    }