@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)); }
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; }
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; } }
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; }  }
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); }
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);  }