### Question:
StringKeyMd5Partitioner implements KafkaMessagePartitioner { public int partition(Object object, List<PartitionInfo> partitions) { int numPartitions = partitions.size(); Preconditions.checkArgument(numPartitions > 0); byte[] key = (byte[]) object; String strKey = new String((byte[]) key); int partitionNum = hf.newHasher(16).putString(strKey, Charsets.UTF_8).hash().asInt() % numPartitions; if (partitionNum < 0) { partitionNum += numPartitions; } LOG.debug(String.format("The partition number for key {} is {}", key, partitionNum)); return partitionNum; }  int partition(Object object, List<PartitionInfo> partitions);  }### Answer:
@Test public void testPartitionWithStringKey() { List<PartitionInfo> partitions = Arrays.asList(new PartitionInfo[50]); String strKey = "399553798187727154"; byte[] key = strKey.getBytes(Charsets.UTF_8); assertEquals(24, partitioner.partition(key, partitions)); }
@Test public void testPartitionWithWrongKeyType() { List<PartitionInfo> partitions = Arrays.asList(new PartitionInfo[50]); Integer intKey = new Integer(1); try { partitioner.partition(new Byte[]{intKey.byteValue()}, partitions); fail("Should throw runtime exception on wrong key type"); } catch (Throwable e) { assertTrue(e instanceof ClassCastException); } }