Java 类org.apache.kafka.clients.producer.Partitioner 实例源码

@Test
public void testPartitionSpread() throws Exception {
    Multiset<Integer> results = TreeMultiset.create();
    Cluster c = Cluster.empty();
    try (Partitioner p = new DefaultPartitioner()) {
        PartitionKeyGenerator pkg = new PartitionKeyGenerator();

        mockPartitions(c);

        for (int i = 0; i < messages; i++) {
            results.add(p.partition("test", null, pkg.next(), null, null, c));
        }

        int expected = messages / partitions;
        double threshold = expected * 0.05;

        for (Multiset.Entry<Integer> e : results.entrySet()) {
            int offBy = Math.abs(e.getCount() - expected);
            assertTrue("Partition " + e.getElement() + " had " + e.getCount() + " elements, expected " + expected + ", threshold is " + threshold,
                    offBy < threshold);
        }
    }
}

private int getProducerPartition(StatEventKey statEventKey, int numPartitions) {
    Partitioner partitioner = new StatEventKeyPartitioner();
    int partition = partitioner.partition(topic, statEventKey, statEventKey.getBytes(), null, null, cluster);
    Assertions.assertThat(partition).isNotNegative();
    Assertions.assertThat(partition).isLessThan(numPartitions);
    return partition;
}