GroupPartitioner { public int getPartitionId(final PartitionKey partitionKey) { int result = 0; for (final PartitionKey key : splitPoints) { if (partitionKey.compareTo(key) < 0) { return result; } result++; } return result; } GroupPartitioner(final String group, final List<PartitionKey> splitPoints); String getGroup(); int getPartitionId(final PartitionKey partitionKey); List<Integer> getPartitionIds(final Object[] partialKey); PartitionKey getIthPartitionKey(final int i); List<PartitionKey> getSplitPoints(); List<Partition> getPartitions(); @Override String toString(); @Override boolean equals(final Object obj); @Override int hashCode();  }
@Test public void shouldCorrectlyAssignPartitionIDs() { final PartitionKey key1 = new PartitionKey(new Object[]{1L, "B"}); final PartitionKey key2 = new PartitionKey(new Object[]{5L, "A"}); final PartitionKey key3 = new PartitionKey(new Object[]{100L, "Z"}); final List<PartitionKey> splitPoints = new ArrayList<>(Arrays.asList(key1, key2, key3)); final GroupPartitioner partitioner = new GroupPartitioner("GROUP", splitPoints); final int partition1 = partitioner.getPartitionId(new PartitionKey(new Object[]{0L, "Z"})); final int partition2 = partitioner.getPartitionId(new PartitionKey(new Object[]{1L, "B"})); final int partition3 = partitioner.getPartitionId(new PartitionKey(new Object[]{1L, "C"})); final int partition4 = partitioner.getPartitionId(new PartitionKey(new Object[]{5L, "A"})); final int partition5 = partitioner.getPartitionId(new PartitionKey(new Object[]{5L, "B"})); final int partition6 = partitioner.getPartitionId(new PartitionKey(new Object[]{100L, "Z"})); final int partition7 = partitioner.getPartitionId(new PartitionKey(new Object[]{100L, "ZZ"})); assertEquals(0, partition1); assertEquals(1, partition2); assertEquals(1, partition3); assertEquals(2, partition4); assertEquals(2, partition5); assertEquals(3, partition6); assertEquals(3, partition7); }