@Test public void testChoosePartitionWithKey() { String key1 = "key1"; String key2 = "key2"; Message<?> msg1 = mock(Message.class); when(msg1.hasKey()).thenReturn(true); when(msg1.getKey()).thenReturn(key1); Message<?> msg2 = mock(Message.class); when(msg2.hasKey()).thenReturn(true); when(msg2.getKey()).thenReturn(key2); SinglePartitionMessageRouterImpl router = new SinglePartitionMessageRouterImpl(1234, HashingScheme.JavaStringHash); TopicMetadataImpl metadata = new TopicMetadataImpl(100); assertEquals(key1.hashCode() % 100, router.choosePartition(msg1, metadata)); assertEquals(key2.hashCode() % 100, router.choosePartition(msg2, metadata)); }
@Override public int choosePartition(Message<?> msg, TopicMetadata metadata) { if (msg.hasKey()) { return signSafeMod(hash.makeHash(msg.getKey()), metadata.numPartitions()); } return partitionIndex; }
SinglePartitionMessageRouterImpl extends MessageRouterBase { @Override public int choosePartition(Message<?> msg, TopicMetadata metadata) { if (msg.hasKey()) { return signSafeMod(hash.makeHash(msg.getKey()), metadata.numPartitions()); } return partitionIndex; } }
SinglePartitionMessageRouterImpl extends MessageRouterBase { @Override public int choosePartition(Message<?> msg, TopicMetadata metadata) { if (msg.hasKey()) { return signSafeMod(hash.makeHash(msg.getKey()), metadata.numPartitions()); } return partitionIndex; } SinglePartitionMessageRouterImpl(int partitionIndex, HashingScheme hashingScheme); }
SinglePartitionMessageRouterImpl extends MessageRouterBase { @Override public int choosePartition(Message<?> msg, TopicMetadata metadata) { if (msg.hasKey()) { return signSafeMod(hash.makeHash(msg.getKey()), metadata.numPartitions()); } return partitionIndex; } SinglePartitionMessageRouterImpl(int partitionIndex, HashingScheme hashingScheme); @Override int choosePartition(Message<?> msg, TopicMetadata metadata); }
SinglePartitionMessageRouterImpl extends MessageRouterBase { @Override public int choosePartition(Message<?> msg, TopicMetadata metadata) { if (msg.hasKey()) { return signSafeMod(hash.makeHash(msg.getKey()), metadata.numPartitions()); } return partitionIndex; } SinglePartitionMessageRouterImpl(int partitionIndex, HashingScheme hashingScheme); @Override int choosePartition(Message<?> msg, TopicMetadata metadata);  }