### Question:
BalancedPartitioner extends Partitioner { @Override public int getPartition(Object key) { int elementIdx = key.hashCode(); if (elementIdx <= (elementsPerPartition + 1) * remainder) { int outputPartition = elementIdx / (elementsPerPartition + 1); if (outputPartition >= numPartitions) { outputPartition = getRandom().nextInt(numPartitions); log.trace("Random partition assigned (1): elementIdx={}, numPartitions={}, elementsPerPartition={}, remainder={}", elementIdx, numPartitions, elementsPerPartition, remainder); } return outputPartition; } else { int numValsInLargerPartitions = remainder * (elementsPerPartition + 1); int idxInSmallerPartitions = elementIdx - numValsInLargerPartitions; int smallPartitionIdx = idxInSmallerPartitions / elementsPerPartition; int outputPartition = remainder + smallPartitionIdx; if (outputPartition >= numPartitions) { outputPartition = getRandom().nextInt(numPartitions); log.trace("Random partition assigned (2): elementIdx={}, numPartitions={}, elementsPerPartition={}, remainder={}", elementIdx, numPartitions, elementsPerPartition, remainder); } return outputPartition; } } BalancedPartitioner(int numPartitions, int elementsPerPartition, int remainder); @Override int numPartitions(); @Override int getPartition(Object key);  }### Answer:
@Test public void balancedPartitionerFirstElements() { BalancedPartitioner bp = new BalancedPartitioner(10, 10, 0); for (int i = 0; i < 10; i++) { int p = bp.getPartition(i); assertEquals("Found wrong partition output " + p + ", not 0", 0, p); } }
@Test public void balancedPartitionerFirstElementsWithRemainder() { BalancedPartitioner bp = new BalancedPartitioner(10, 10, 1); for (int i = 0; i < 10; i++) { int p = bp.getPartition(i); assertEquals("Found wrong partition output " + p + ", not 0", 0, p); } }
@Test public void balancedPartitionerDoesBalance() { BalancedPartitioner bp = new BalancedPartitioner(10, 10, 0); int[] countPerPartition = new int[10]; for (int i = 0; i < 10 * 10; i++) { int p = bp.getPartition(i); countPerPartition[p] += 1; } for (int i = 0; i < 10; i++) { assertEquals(10, countPerPartition[i]); } }
@Test public void balancedPartitionerDoesBalanceWithRemainder() { BalancedPartitioner bp = new BalancedPartitioner(10, 10, 7); int[] countPerPartition = new int[10]; for (int i = 0; i < 10 * 10 + 7; i++) { int p = bp.getPartition(i); countPerPartition[p] += 1; } for (int i = 0; i < 10; i++) { if (i < 7) assertEquals(10 + 1, countPerPartition[i]); else assertEquals(10, countPerPartition[i]); } }