KafkaPartitioner implements Partitioner { public static int getPartition(Object key, int numPartitions) { if (numPartitions <= 1) return 0; String keyStr; if (key == null || StrUtils.isEmpty(keyStr = key.toString()) || keyStr.length() > 1_000) return r.nextInt(numPartitions); try { if (StrUtils.isNumber(keyStr)) return (int) Math.abs(Long.parseLong(keyStr) % numPartitions); } catch (Exception e) { _log.error(ExceptionUtils.errorInfo(e)); } return Math.abs(key.hashCode() % numPartitions); } KafkaPartitioner(); static int getPartition(Object key, int numPartitions); @Override int partition(Object key, int numPartitions);  }
@Test public void testSendMessage() { assertEquals(0, KafkaPartitioner.getPartition(null, 0)); assertEquals(0, KafkaPartitioner.getPartition(null, 1)); assertTrue(KafkaPartitioner.getPartition(null, 2) < 2); assertTrue(KafkaPartitioner.getPartition("1", 2) < 2); assertEquals(11, KafkaPartitioner.getPartition("11", 24)); assertEquals(0, KafkaPartitioner.getPartition("24", 24)); }