KafkaUtil { public static Integer getIntegerPartitionKey(OutgoingMessageEnvelope envelope, List<PartitionInfo> partitions) { int numPartitions = partitions.size(); return abs(envelope.getPartitionKey().hashCode()) % numPartitions; }  static String getCheckpointTopic(String jobName, String jobId, Config config); static Integer getIntegerPartitionKey(OutgoingMessageEnvelope envelope, List<PartitionInfo> partitions); static SystemStreamPartition toSystemStreamPartition(String systemName, TopicPartition topicPartition);  }
@Test public void testGetIntegerPartitionKey() { SystemStream systemStream = new SystemStream("system", "stream"); int partitionKeyInt = 10; OutgoingMessageEnvelope outgoingMessageEnvelope = new OutgoingMessageEnvelope(systemStream, partitionKeyInt, "key", "message"); assertEquals(0, KafkaUtil.getIntegerPartitionKey(outgoingMessageEnvelope, partitionInfoList(1)).intValue()); assertEquals(Integer.hashCode(partitionKeyInt) % 3, KafkaUtil.getIntegerPartitionKey(outgoingMessageEnvelope, partitionInfoList(3)).intValue()); assertEquals(Integer.hashCode(partitionKeyInt) % 8, KafkaUtil.getIntegerPartitionKey(outgoingMessageEnvelope, partitionInfoList(8)).intValue()); partitionKeyInt = -10; outgoingMessageEnvelope = new OutgoingMessageEnvelope(systemStream, partitionKeyInt, "key", "message"); assertEquals(0, KafkaUtil.getIntegerPartitionKey(outgoingMessageEnvelope, partitionInfoList(1)).intValue()); assertEquals(Integer.hashCode(Math.abs(partitionKeyInt)) % 3, KafkaUtil.getIntegerPartitionKey(outgoingMessageEnvelope, partitionInfoList(3)).intValue()); assertEquals(Integer.hashCode(Math.abs(partitionKeyInt)) % 8, KafkaUtil.getIntegerPartitionKey(outgoingMessageEnvelope, partitionInfoList(8)).intValue()); partitionKeyInt = Integer.MIN_VALUE; outgoingMessageEnvelope = new OutgoingMessageEnvelope(systemStream, partitionKeyInt, "key", "message"); assertEquals(0, KafkaUtil.getIntegerPartitionKey(outgoingMessageEnvelope, partitionInfoList(1)).intValue()); assertEquals(0, KafkaUtil.getIntegerPartitionKey(outgoingMessageEnvelope, partitionInfoList(3)).intValue()); assertEquals(0, KafkaUtil.getIntegerPartitionKey(outgoingMessageEnvelope, partitionInfoList(8)).intValue()); String partitionKeyString = "abc"; outgoingMessageEnvelope = new OutgoingMessageEnvelope(systemStream, partitionKeyString, "key", "message"); assertEquals(0, KafkaUtil.getIntegerPartitionKey(outgoingMessageEnvelope, partitionInfoList(1)).intValue()); assertEquals(Math.abs(partitionKeyString.hashCode()) % 3, KafkaUtil.getIntegerPartitionKey(outgoingMessageEnvelope, partitionInfoList(3)).intValue()); assertEquals(Math.abs(partitionKeyString.hashCode()) % 8, KafkaUtil.getIntegerPartitionKey(outgoingMessageEnvelope, partitionInfoList(8)).intValue()); }