### Question:
Crc32ByteArrayPartitioner implements KafkaMessagePartitioner { public int partition(Object object, List<PartitionInfo> partitions) { int numOfPartitions = partitions.size(); Preconditions.checkArgument(numOfPartitions > 0); byte[] key = (byte[]) object; if (object == null) { return Math.abs(random.nextInt() % numOfPartitions); } else { int hash = hashFunction.hashBytes(key).asInt(); LOG.debug("Crc32ByteArrayPartitioner.partition producer hash: {} on key: {}", hash, Arrays.toString(key)); return Math.abs(Math.abs(hash) % numOfPartitions); } } Crc32ByteArrayPartitioner(); int partition(Object object, List<PartitionInfo> partitions);  }### Answer:
@Test public void testCrc32ByteArrayPartitionerTestWithByteArray() { List<PartitionInfo> partitions = Arrays.asList(new PartitionInfo[60]); assertEquals(40, partitioner.partition(new byte[]{0, 80, 64, 36, 43, 101, 22, 13}, partitions)); }
@Test public void testCrc32ByteArrayPartitionerTestWithByteArrayWithIntMinHash() { List<PartitionInfo> partitions = Arrays.asList(new PartitionInfo[60]); assertEquals(8, partitioner.partition(new byte[]{0, 32, 69, -4, -59, 25, -32, 22}, partitions)); }
@Test public void testCrc32ByteArrayPartitionerTestWithWrongKeyType() { List<PartitionInfo> partitions = Arrays.asList(new PartitionInfo[60]); try { partitioner.partition(new Byte[]{1, 2}, partitions); fail("Should throw runtime exception on wrong key type"); } catch (Throwable e) { assertTrue(e instanceof ClassCastException); } }